/*-------------------------------------------------------------------------
 *
 * nodeHash.c
 *	  为 hashjoin 进行关系散列的例程
 *
 * Portions Copyright (c) 1996-2022, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994, Regents of the University of California
 *
 *
 * IDENTIFICATION
 *	  src/backend/executor/nodeHash.c
 *
 * 有关 nodeHashjoin.c 中并行性的说明。
 *
 *-------------------------------------------------------------------------
 */
/*
 * 接口例程
 *		MultiExecHash	- 生成关系的内存哈希表
 *		ExecInitHash	- 初始化节点和子节点
 *		ExecEndHash		- 关闭节点和子节点
 */

#include "postgres.h"

#include <math.h>
#include <limits.h>

#include "access/htup_details.h"
#include "access/parallel.h"
#include "catalog/pg_statistic.h"
#include "commands/tablespace.h"
#include "executor/execdebug.h"
#include "executor/hashjoin.h"
#include "executor/nodeHash.h"
#include "executor/nodeHashjoin.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "port/atomics.h"
#include "port/pg_bitutils.h"
#include "utils/dynahash.h"
#include "utils/guc.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/syscache.h"

static void fc_ExecHashIncreaseNumBatches(HashJoinTable hashtable);
static void fc_ExecHashIncreaseNumBuckets(HashJoinTable hashtable);
static void fc_ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable);
static void fc_ExecParallelHashIncreaseNumBuckets(HashJoinTable hashtable);
static void fc_ExecHashBuildSkewHash(HashJoinTable hashtable, Hash *fc_node,
								  int fc_mcvsToUse);
static void fc_ExecHashSkewTableInsert(HashJoinTable hashtable,
									TupleTableSlot *fc_slot,
									uint32 fc_hashvalue,
									int fc_bucketNumber);
static void fc_ExecHashRemoveNextSkewBucket(HashJoinTable hashtable);

static void *fc_dense_alloc(HashJoinTable hashtable, Size fc_size);
static HashJoinTuple fc_ExecParallelHashTupleAlloc(HashJoinTable hashtable,
												size_t fc_size,
												dsa_pointer *fc_shared);
static void fc_MultiExecPrivateHash(HashState *fc_node);
static void fc_MultiExecParallelHash(HashState *fc_node);
static inline HashJoinTuple fc_ExecParallelHashFirstTuple(HashJoinTable fc_table,
													   int fc_bucketno);
static inline HashJoinTuple fc_ExecParallelHashNextTuple(HashJoinTable fc_table,
													  HashJoinTuple fc_tuple);
static inline void fc_ExecParallelHashPushTuple(dsa_pointer_atomic *fc_head,
											 HashJoinTuple fc_tuple,
											 dsa_pointer fc_tuple_shared);
static void fc_ExecParallelHashJoinSetUpBatches(HashJoinTable hashtable, int fc_nbatch);
static void fc_ExecParallelHashEnsureBatchAccessors(HashJoinTable hashtable);
static void fc_ExecParallelHashRepartitionFirst(HashJoinTable hashtable);
static void fc_ExecParallelHashRepartitionRest(HashJoinTable hashtable);
static HashMemoryChunk fc_ExecParallelHashPopChunkQueue(HashJoinTable fc_table,
													 dsa_pointer *fc_shared);
static bool fc_ExecParallelHashTuplePrealloc(HashJoinTable hashtable,
										  int fc_batchno,
										  size_t fc_size);
static void fc_ExecParallelHashMergeCounters(HashJoinTable hashtable);
static void fc_ExecParallelHashCloseBatchAccessors(HashJoinTable hashtable);


/* ----------------------------------------------------------------
 *		ExecHash
 *
 *		形式合规的存根
 * ----------------------------------------------------------------
 */
static TupleTableSlot * fc_ExecHash(PlanState *fc_pstate)
{
	elog(ERROR, "Hash node does not support ExecProcNode call convention");
	return NULL;
}

/* ----------------------------------------------------------------
 *		MultiExecHash
 *
 *		为哈希连接构建哈希表，如果需要多个批次则进行分区。
 * ----------------------------------------------------------------
 */
Node * MultiExecHash(HashState *fc_node)
{
	/* 必须提供我们自己的仪器支持 */
	if (fc_node->ps.instrument)
		InstrStartNode(fc_node->ps.instrument);

	if (fc_node->parallel_state != NULL)
		fc_MultiExecParallelHash(fc_node);
	else
		fc_MultiExecPrivateHash(fc_node);

	/* 必须提供我们自己的仪器支持 */
	if (fc_node->ps.instrument)
		InstrStopNode(fc_node->ps.instrument, fc_node->hashtable->partialTuples);

	/*
	 * 我们不直接返回哈希表，因为它不是Node的子类型，因此会违反MultiExecProcNode API。相反，我们的父Hashjoin节点应该知道如何从我们的节点状态中提取它。虽然不太好，但实际上不值得清理，因为Hashjoin对Hash的了解远不止这些。
	 */
	return NULL;
}

/* ----------------------------------------------------------------
 *		MultiExecPrivateHash
 *
 *		并行无关版本，构建后端私有哈希表和（如果需要）批处理文件。
 * ----------------------------------------------------------------
 */
static void fc_MultiExecPrivateHash(HashState *fc_node)
{
	PlanState  *fc_outerNode;
	List	   *fc_hashkeys;
	HashJoinTable hashtable;
	TupleTableSlot *fc_slot;
	ExprContext *fc_econtext;
	uint32		fc_hashvalue;

	/*
	 * 从节点获取状态信息
	 */
	fc_outerNode = outerPlanState(fc_node);
	hashtable = fc_node->hashtable;

	/*
	 * 设置表达式上下文
	 */
	fc_hashkeys = fc_node->hashkeys;
	fc_econtext = fc_node->ps.ps_ExprContext;

	/*
	 * 从Hash节点下方的节点获取所有元组并插入到哈希表（或临时文件）中。
	 */
	for (;;)
	{
		fc_slot = ExecProcNode(fc_outerNode);
		if (TupIsNull(fc_slot))
			break;
		/* 我们必须计算哈希值 */
		fc_econtext->ecxt_outertuple = fc_slot;
		if (ExecHashGetHashValue(hashtable, fc_econtext, fc_hashkeys,
								 false, hashtable->keepNulls,
								 &fc_hashvalue))
		{
			int			fc_bucketNumber;

			fc_bucketNumber = ExecHashGetSkewBucket(hashtable, fc_hashvalue);
			if (fc_bucketNumber != INVALID_SKEW_BUCKET_NO)
			{
				/* 这是一个偏斜元组，因此将其放入该哈希表中 */
				fc_ExecHashSkewTableInsert(hashtable, fc_slot, fc_hashvalue,
										fc_bucketNumber);
				hashtable->skewTuples += 1;
			}
			else
			{
				/* 不受偏斜优化影响，因此正常插入 */
				ExecHashTableInsert(hashtable, fc_slot, fc_hashvalue);
			}
			hashtable->totalTuples += 1;
		}
	}

	/* 如果需要，调整哈希表大小（超出NTUP_PER_BUCKET） */
	if (hashtable->nbuckets != hashtable->nbuckets_optimal)
		fc_ExecHashIncreaseNumBuckets(hashtable);

	/* 在spaceUsed中计算桶（在EXPLAIN ANALYZE中报告） */
	hashtable->spaceUsed += hashtable->nbuckets * sizeof(HashJoinTuple);
	if (hashtable->spaceUsed > hashtable->spacePeak)
		hashtable->spacePeak = hashtable->spaceUsed;

	hashtable->partialTuples = hashtable->totalTuples;
}

/* ----------------------------------------------------------------
 *		MultiExecParallelHash
 *
 *		并行感知版本，构建共享哈希表和（如有必要）批处理文件，使用一组合作后端的共同努力。
 * ----------------------------------------------------------------
 */
static void fc_MultiExecParallelHash(HashState *fc_node)
{
	ParallelHashJoinState *fc_pstate;
	PlanState  *fc_outerNode;
	List	   *fc_hashkeys;
	HashJoinTable hashtable;
	TupleTableSlot *fc_slot;
	ExprContext *fc_econtext;
	uint32		fc_hashvalue;
	Barrier    *fc_build_barrier;
	int			fc_i;

	/*
	 * 从节点获取状态信息
	 */
	fc_outerNode = outerPlanState(fc_node);
	hashtable = fc_node->hashtable;

	/*
	 * 设置表达式上下文
	 */
	fc_hashkeys = fc_node->hashkeys;
	fc_econtext = fc_node->ps.ps_ExprContext;

	/*
	 * 同步并行哈希表的构建。在这个阶段，我们知道共享哈希表已经或正在被ExecHashTableCreate()设置，但我们不知道我们的同伴是否已经从那里返回，或者是否在MultiExecParallelHash()中，以及他们目前进展到哪一步。为了找出这一点，我们检查build_barrier阶段，然后跳到构建算法中的正确步骤。
	 */
	fc_pstate = hashtable->parallel_state;
	fc_build_barrier = &fc_pstate->build_barrier;
	Assert(BarrierPhase(fc_build_barrier) >= PHJ_BUILD_ALLOCATING);
	switch (BarrierPhase(fc_build_barrier))
	{
		case PHJ_BUILD_ALLOCATING:

			/*
			 * 我可能刚在ExecHashTableCreate()中分配了初始哈希表，或者其他人正在做这一操作。无论哪种方式，等待所有人到达这里，以便我们可以继续。
			 */
			BarrierArriveAndWait(fc_build_barrier, WAIT_EVENT_HASH_BUILD_ALLOCATE);
			/* Fall through. */

		case PHJ_BUILD_HASHING_INNER:

			/*
			 * 现在是开始哈希的时候，或者如果我们刚刚到达这里，那么哈希已经在进行中，
			 * 因此加入这项工作。在哈希时，我们必须准备好随时帮助增加批次或桶的数量，
			 * 如果我们在这里到达时这项工作已经开始，我们要立即帮助完成这项工作，
			 * 以确保可以安全地访问下面的批次和桶。
			 */
			if (PHJ_GROW_BATCHES_PHASE(BarrierAttach(&fc_pstate->grow_batches_barrier)) !=
				PHJ_GROW_BATCHES_ELECTING)
				fc_ExecParallelHashIncreaseNumBatches(hashtable);
			if (PHJ_GROW_BUCKETS_PHASE(BarrierAttach(&fc_pstate->grow_buckets_barrier)) !=
				PHJ_GROW_BUCKETS_ELECTING)
				fc_ExecParallelHashIncreaseNumBuckets(hashtable);
			fc_ExecParallelHashEnsureBatchAccessors(hashtable);
			ExecParallelHashTableSetCurrentBatch(hashtable, 0);
			for (;;)
			{
				fc_slot = ExecProcNode(fc_outerNode);
				if (TupIsNull(fc_slot))
					break;
				fc_econtext->ecxt_outertuple = fc_slot;
				if (ExecHashGetHashValue(hashtable, fc_econtext, fc_hashkeys,
										 false, hashtable->keepNulls,
										 &fc_hashvalue))
					ExecParallelHashTableInsert(hashtable, fc_slot, fc_hashvalue);
				hashtable->partialTuples++;
			}

			/*
			 * 确保我们写入磁盘的任何元组在任何人尝试加载它们之前对其他人可见。
			 */
			for (fc_i = 0; fc_i < hashtable->nbatch; ++fc_i)
				sts_end_write(hashtable->batches[fc_i].inner_tuples);

			/*
			 * 更新共享计数器。我们需要准确的元组总数来控制空表优化。
			 */
			fc_ExecParallelHashMergeCounters(hashtable);

			BarrierDetach(&fc_pstate->grow_buckets_barrier);
			BarrierDetach(&fc_pstate->grow_batches_barrier);

			/*
			 * 等待每个人完成构建和刷新文件及计数器。
			 */
			if (BarrierArriveAndWait(fc_build_barrier,
									 WAIT_EVENT_HASH_BUILD_HASH_INNER))
			{
				/*
				 * 选举一个后端来禁用任何进一步的增长。批次现在是固定的。
				 * 在构建它们时，我们确保它们在后续加载时能够适应我们的内存预算（或者我们尝试这样做但放弃了，因为我们检测到极端的偏斜）。
				 */
				fc_pstate->growth = PHJ_GROWTH_DISABLED;
			}
	}

	/*
	 * 我们还没有附加到一个批次。我们都同意内元组的维度和数量（用于空表优化）。
	 */
	hashtable->curbatch = -1;
	hashtable->nbuckets = fc_pstate->nbuckets;
	hashtable->log2_nbuckets = my_log2(hashtable->nbuckets);
	hashtable->totalTuples = fc_pstate->total_tuples;

	/*
	 * 除非我们完全完成并且批次状态已释放，否则确保我们有访问器。
	 */
	if (BarrierPhase(fc_build_barrier) < PHJ_BUILD_DONE)
		fc_ExecParallelHashEnsureBatchAccessors(hashtable);

	/*
	 * 下一个同步点是在 ExecHashJoin 的 HJ_BUILD_HASHTABLE 案例中，这将使构建阶段进入 PHJ_BUILD_RUNNING（如果尚未在那里的话）。
	 */
	Assert(BarrierPhase(fc_build_barrier) == PHJ_BUILD_HASHING_OUTER ||
		   BarrierPhase(fc_build_barrier) == PHJ_BUILD_RUNNING ||
		   BarrierPhase(fc_build_barrier) == PHJ_BUILD_DONE);
}

/* ----------------------------------------------------------------
 *		ExecInitHash
 *
 *		Hash 节点的初始化例程
 * ----------------------------------------------------------------
 */
HashState *
ExecInitHash(Hash *fc_node, EState *fc_estate, int fc_eflags)
{
	HashState  *fc_hashstate;

	/* 检查不支持的标志 */
	Assert(!(fc_eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK)));

	/*
	 * 创建状态结构
	 */
	fc_hashstate = makeNode(HashState);
	fc_hashstate->ps.plan = (Plan *) fc_node;
	fc_hashstate->ps.state = fc_estate;
	fc_hashstate->ps.ExecProcNode = fc_ExecHash;
	fc_hashstate->hashtable = NULL;
	fc_hashstate->hashkeys = NIL;	/* 将由父级 HashJoin 设置 */

	/*
	 * 杂项初始化
	 *
	 * 为节点创建表达式上下文
	 */
	ExecAssignExprContext(fc_estate, &fc_hashstate->ps);

	/*
	 * 初始化子节点
	 */
	outerPlanState(fc_hashstate) = ExecInitNode(outerPlan(fc_node), fc_estate, fc_eflags);

	/*
	 * 初始化我们的结果槽和类型。无需构建投影，因为该节点不执行投影。
	 */
	ExecInitResultTupleSlotTL(&fc_hashstate->ps, &TTSOpsMinimalTuple);
	fc_hashstate->ps.ps_ProjInfo = NULL;

	/*
	 * 初始化子表达式
	 */
	Assert(fc_node->plan.qual == NIL);
	fc_hashstate->hashkeys =
		ExecInitExprList(fc_node->hashkeys, (PlanState *) fc_hashstate);

	return fc_hashstate;
}

/* ---------------------------------------------------------------
 *		ExecEndHash
 *
 *		Hash 节点的清理例程
 * ----------------------------------------------------------------
 */
void ExecEndHash(HashState *fc_node)
{
	PlanState  *fc_outerPlan;

	/*
	 * 释放 exprcontext
	 */
	ExecFreeExprContext(&fc_node->ps);

	/*
	 * 关闭子计划
	 */
	fc_outerPlan = outerPlanState(fc_node);
	ExecEndNode(fc_outerPlan);
}


/* ----------------------------------------------------------------
 *		ExecHashTableCreate
 *
 *		为 hashjoin 创建一个空的哈希表数据结构。
 * ----------------------------------------------------------------
 */
HashJoinTable
ExecHashTableCreate(HashState *fc_state, List *fc_hashOperators, List *fc_hashCollations, bool fc_keepNulls)
{
	Hash	   *fc_node;
	HashJoinTable hashtable;
	Plan	   *fc_outerNode;
	size_t		fc_space_allowed;
	int			fc_nbuckets;
	int			fc_nbatch;
	double		fc_rows;
	int			fc_num_skew_mcvs;
	int			fc_log2_nbuckets;
	int			fc_nkeys;
	int			fc_i;
	ListCell   *fc_ho;
	ListCell   *fc_hc;
	MemoryContext fc_oldcxt;

	/*
	 * 获取有关要哈希的关系大小的信息（这是该节点的“外部”子树，但 hashjoin 的内部关系）。
	 * 计算哈希表的适当大小。
	 */
	fc_node = (Hash *) fc_state->ps.plan;
	fc_outerNode = outerPlan(fc_node);

	/*
	 * 如果这是一个带有部分计划的共享哈希表，则我们无法使用 outerNode->plan_rows 来估算其大小。
	 * 我们需要对所有部分计划的副本中行的总数进行估算。
	 */
	fc_rows = fc_node->plan.parallel_aware ? fc_node->rows_total : fc_outerNode->plan_rows;

	ExecChooseHashTableSize(fc_rows, fc_outerNode->plan_width,
							OidIsValid(fc_node->skewTable),
							fc_state->parallel_state != NULL,
							fc_state->parallel_state != NULL ?
							fc_state->parallel_state->nparticipants - 1 : 0,
							&fc_space_allowed,
							&fc_nbuckets, &fc_nbatch, &fc_num_skew_mcvs);

	/* nbuckets 必须是 2 的幂 */
	fc_log2_nbuckets = my_log2(fc_nbuckets);
	Assert(fc_nbuckets == (1 << fc_log2_nbuckets));

	/*
	 * 初始化哈希表控制块。
	 *
	 * 哈希表控制块仅从执行器的每查询内存上下文中分配。
	 * 其他所有内容应保留在子哈希上下文或批次上下文中。
	 */
	hashtable = (HashJoinTable) palloc(sizeof(HashJoinTableData));
	hashtable->nbuckets = fc_nbuckets;
	hashtable->nbuckets_original = fc_nbuckets;
	hashtable->nbuckets_optimal = fc_nbuckets;
	hashtable->log2_nbuckets = fc_log2_nbuckets;
	hashtable->log2_nbuckets_optimal = fc_log2_nbuckets;
	hashtable->buckets.unshared = NULL;
	hashtable->keepNulls = fc_keepNulls;
	hashtable->skewEnabled = false;
	hashtable->skewBucket = NULL;
	hashtable->skewBucketLen = 0;
	hashtable->nSkewBuckets = 0;
	hashtable->skewBucketNums = NULL;
	hashtable->nbatch = fc_nbatch;
	hashtable->curbatch = 0;
	hashtable->nbatch_original = fc_nbatch;
	hashtable->nbatch_outstart = fc_nbatch;
	hashtable->growEnabled = true;
	hashtable->totalTuples = 0;
	hashtable->partialTuples = 0;
	hashtable->skewTuples = 0;
	hashtable->innerBatchFile = NULL;
	hashtable->outerBatchFile = NULL;
	hashtable->spaceUsed = 0;
	hashtable->spacePeak = 0;
	hashtable->spaceAllowed = fc_space_allowed;
	hashtable->spaceUsedSkew = 0;
	hashtable->spaceAllowedSkew =
		hashtable->spaceAllowed * SKEW_HASH_MEM_PERCENT / 100;
	hashtable->chunks = NULL;
	hashtable->current_chunk = NULL;
	hashtable->parallel_state = fc_state->parallel_state;
	hashtable->area = fc_state->ps.state->es_query_dsa;
	hashtable->batches = NULL;

#ifdef HJDEBUG
	printf("Hashjoin %p: initial nbatch = %d, nbuckets = %d\n",
		   hashtable, fc_nbatch, fc_nbuckets);
#endif

	/*
	 * 创建临时内存上下文，以保持哈希表的工作存储。
	 * 请参见 executor/hashjoin.h 中的说明。
	 */
	hashtable->hashCxt = AllocSetContextCreate(CurrentMemoryContext,
											   "HashTableContext",
											   ALLOCSET_DEFAULT_SIZES);

	hashtable->batchCxt = AllocSetContextCreate(hashtable->hashCxt,
												"HashBatchContext",
												ALLOCSET_DEFAULT_SIZES);

	/* 分配将在 hashjoin 生命周期中存在的数据 */

	fc_oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);

	/*
	 * 获取有关每个哈希键的哈希函数的信息。还需记住连接运算符是否严格。
	 */
	fc_nkeys = list_length(fc_hashOperators);
	hashtable->outer_hashfunctions =
		(FmgrInfo *) palloc(fc_nkeys * sizeof(FmgrInfo));
	hashtable->inner_hashfunctions =
		(FmgrInfo *) palloc(fc_nkeys * sizeof(FmgrInfo));
	hashtable->hashStrict = (bool *) palloc(fc_nkeys * sizeof(bool));
	hashtable->collations = (Oid *) palloc(fc_nkeys * sizeof(Oid));
	fc_i = 0;
	forboth(fc_ho, fc_hashOperators, fc_hc, fc_hashCollations)
	{
		Oid			fc_hashop = lfirst_oid(fc_ho);
		Oid			fc_left_hashfn;
		Oid			fc_right_hashfn;

		if (!get_op_hash_functions(fc_hashop, &fc_left_hashfn, &fc_right_hashfn))
			elog(ERROR, "could not find hash function for hash operator %u",
				 fc_hashop);
		fmgr_info(fc_left_hashfn, &hashtable->outer_hashfunctions[fc_i]);
		fmgr_info(fc_right_hashfn, &hashtable->inner_hashfunctions[fc_i]);
		hashtable->hashStrict[fc_i] = op_strict(fc_hashop);
		hashtable->collations[fc_i] = lfirst_oid(fc_hc);
		fc_i++;
	}

	if (fc_nbatch > 1 && hashtable->parallel_state == NULL)
	{
		/*
		 * 在 hashCxt 中分配并初始化文件数组（并不需要用于使用共享元组存储而不是原始文件的并行情况）
		 */
		hashtable->innerBatchFile = (BufFile **)
			palloc0(fc_nbatch * sizeof(BufFile *));
		hashtable->outerBatchFile = (BufFile **)
			palloc0(fc_nbatch * sizeof(BufFile *));
		/* 文件在需要之前不会打开... */
		/* ... 但确保我们已建立临时表空间 */
		PrepareTempTablespaces();
	}

	MemoryContextSwitchTo(fc_oldcxt);

	if (hashtable->parallel_state)
	{
		ParallelHashJoinState *fc_pstate = hashtable->parallel_state;
		Barrier    *fc_build_barrier;

		/*
		 * 附加到构建屏障。相应的分离操作在 ExecHashTableDetach 中。
		 * 注意，我们尚未附加到批次 0 的 batch_barrier。我们稍后会附加并将其从 PHJ_BATCH_PROBING 阶段开始，因为批次 0 是预先分配的，
		 * 然后在哈希时加载（标准混合哈希连接算法），我们将使用 build_barrier 协调。
		 */
		fc_build_barrier = &fc_pstate->build_barrier;
		BarrierAttach(fc_build_barrier);

		/*
		 * 到目前为止，我们不知道是否有其他参与者，如果有，他们正处于哪个阶段。
		 * 此时我们唯一关心的是是否有人已经创建了 SharedHashJoinBatch 对象和批次 0 的哈希表。
		 * 如果有必要，将选举一个后端来完成这项工作。
		 */
		if (BarrierPhase(fc_build_barrier) == PHJ_BUILD_ELECTING &&
			BarrierArriveAndWait(fc_build_barrier, WAIT_EVENT_HASH_BUILD_ELECT))
		{
			fc_pstate->nbatch = fc_nbatch;
			fc_pstate->space_allowed = fc_space_allowed;
			fc_pstate->growth = PHJ_GROWTH_OK;

			/* 设置共享状态以协调批次。 */
			fc_ExecParallelHashJoinSetUpBatches(hashtable, fc_nbatch);

			/*
			 * 提前分配批次 0 的哈希表，以便在哈希时可以直接加载它。
			 */
			fc_pstate->nbuckets = fc_nbuckets;
			ExecParallelHashTableAlloc(hashtable, 0);
		}

		/*
		 * 下一个并行哈希同步点在 MultiExecParallelHash() 中，这将使其一直进展到 PHJ_BUILD_RUNNING。
		 * 从现在到那时，调用者不得从此执行节点返回控制权。
		 */
	}
	else
	{
		/*
		 * 准备第一个扫描空间分配的上下文；在其中分配哈希桶数组，并将每个桶设置为“空”。
		 */
		MemoryContextSwitchTo(hashtable->batchCxt);

		hashtable->buckets.unshared = (HashJoinTuple *)
			palloc0(fc_nbuckets * sizeof(HashJoinTuple));

		/*
		 * 如果可能且需要多个批次，则进行倾斜优化的设置。 （在单批次联接中，没有必要。）
		 */
		if (fc_nbatch > 1)
			fc_ExecHashBuildSkewHash(hashtable, fc_node, fc_num_skew_mcvs);

		MemoryContextSwitchTo(fc_oldcxt);
	}

	return hashtable;
}


/*
 * 根据要哈希的关系的估计大小（行数和平均行宽）计算哈希表的适当大小。
 *
 * 这被导出，以便规划器的成本大小可以使用它。
 */

/* 目标桶加载（每个桶的元组数） */
#define NTUP_PER_BUCKET			1

void ExecChooseHashTableSize(double fc_ntuples, int fc_tupwidth, bool fc_useskew,
						bool fc_try_combined_hash_mem,
						int fc_parallel_workers,
						size_t *fc_space_allowed,
						int *fc_numbuckets,
						int *fc_numbatches,
						int *fc_num_skew_mcvs)
{
	int			fc_tupsize;
	double		fc_inner_rel_bytes;
	size_t		fc_hash_table_bytes;
	size_t		fc_bucket_bytes;
	size_t		fc_max_pointers;
	int			fc_nbatch = 1;
	int			fc_nbuckets;
	double		fc_dbuckets;

	/* 如果没有信息，强制一个合理的关系大小 */
	if (fc_ntuples <= 0.0)
		fc_ntuples = 1000.0;

	/*
	 * 根据哈希表中元组的占用空间估计元组大小... 注意，这
	 * 不允许任何palloc开销。  对空间使用的操作
	 * 也不计算palloc开销。
	 */
	fc_tupsize = HJTUPLE_OVERHEAD +
		MAXALIGN(SizeofMinimalTupleHeader) +
		MAXALIGN(fc_tupwidth);
	fc_inner_rel_bytes = fc_ntuples * fc_tupsize;

	/*
	 * 从GUC计算内存中哈希表的大小限制。
	 */
	fc_hash_table_bytes = get_hash_memory_limit();

	/*
	 * 并行哈希尝试利用所有工作者的组合hash_mem
	 * 来避免需要批处理。 如果这行不通，它将回退到每个工作者的hash_mem
	 * 并尝试并行处理批次。
	 */
	if (fc_try_combined_hash_mem)
	{
		/* 小心，这可能会溢出size_t */
		double		fc_newlimit;

		fc_newlimit = (double) fc_hash_table_bytes * (double) (fc_parallel_workers + 1);
		fc_newlimit = Min(fc_newlimit, (double) SIZE_MAX);
		fc_hash_table_bytes = (size_t) fc_newlimit;
	}

	*fc_space_allowed = fc_hash_table_bytes;

	/*
	 * 如果倾斜优化是可能的，估计将在允许的内存中适合的倾斜桶数量，
	 * 并相应减少主要哈希表的假定可用空间。
	 *
	 * 我们做出乐观的假设，假设每个倾斜桶将包含一个内部关系元组。 如果结果证明
	 * 这太少，我们将在运行时通过减少倾斜桶的数量来恢复。
	 *
	 * hashtable->skewBucket将具有最多8倍于
	 * 我们允许的MCV数量的HashSkewBucket指针，因为
	 * ExecHashBuildSkewHash将向上取整到下一个2的幂，然后乘以4以减少
	 * 碰撞。
	 */
	if (fc_useskew)
	{
		size_t		fc_bytes_per_mcv;
		size_t		fc_skew_mcvs;

		/*----------
		 * 计算我们可以在hash_table_bytes中容纳的MCV数量
		 *
		 * 除数为：
		 * 哈希元组的大小 +
		 * 每个MCV的skewBucket[]的最坏情况大小 +
		 * skewBucketNums[]条目的大小 +
		 * 倾斜桶结构本身的大小
		 *----------
		 */
		fc_bytes_per_mcv = fc_tupsize +
			(8 * sizeof(HashSkewBucket *)) +
			sizeof(int) +
			SKEW_BUCKET_OVERHEAD;
		fc_skew_mcvs = fc_hash_table_bytes / fc_bytes_per_mcv;

		/*
		 * 现在按SKEW_HASH_MEM_PERCENT缩放（我们以这种顺序进行
		 * 以免在乘法中担心size_t溢出）
		 */
		fc_skew_mcvs = (fc_skew_mcvs * SKEW_HASH_MEM_PERCENT) / 100;

		/* 现在限制为整数范围 */
		fc_skew_mcvs = Min(fc_skew_mcvs, INT_MAX);

		*fc_num_skew_mcvs = (int) fc_skew_mcvs;

		/* 减少hash_table_bytes以满足倾斜表的需求 */
		if (fc_skew_mcvs > 0)
			fc_hash_table_bytes -= fc_skew_mcvs * fc_bytes_per_mcv;
	}
	else
		*fc_num_skew_mcvs = 0;

	/*
	 * 将nbuckets设置为在内存填满时实现平均每个桶NTUP_PER_BUCKET加载，
	 * 假设为单个批次；但限制该值，以便我们尝试分配的指针数组不超过
	 * hash_table_bytes或MaxAllocSize。
	 *
	 * 请注意，nbuckets和nbatch都必须是2的幂，以使
	 * ExecHashGetBucketAndBatch快速。
	 */
	fc_max_pointers = fc_hash_table_bytes / sizeof(HashJoinTuple);
	fc_max_pointers = Min(fc_max_pointers, MaxAllocSize / sizeof(HashJoinTuple));
	/* 如果max_pointers不是2的幂，必须向下舍入到一个 */
	fc_max_pointers = pg_prevpower2_size_t(fc_max_pointers);

	/* 还要确保我们避免nbatch和nbuckets中的整数溢出 */
	/* （鉴于当前的MaxAllocSize，这一步是多余的） */
	fc_max_pointers = Min(fc_max_pointers, INT_MAX / 2 + 1);

	fc_dbuckets = ceil(fc_ntuples / NTUP_PER_BUCKET);
	fc_dbuckets = Min(fc_dbuckets, fc_max_pointers);
	fc_nbuckets = (int) fc_dbuckets;
	/* 不要让nbuckets真的很小，尽管... */
	fc_nbuckets = Max(fc_nbuckets, 1024);
	/* ...并强制其为2的幂。 */
	fc_nbuckets = pg_nextpower2_32(fc_nbuckets);

	/*
	 * 如果没有足够的空间存储预计的元组数量和所需的桶头文件，
	 * 我们将需要多个批次。
	 */
	fc_bucket_bytes = sizeof(HashJoinTuple) * fc_nbuckets;
	if (fc_inner_rel_bytes + fc_bucket_bytes > fc_hash_table_bytes)
	{
		/* 我们将需要多个批次 */
		size_t		fc_sbuckets;
		double		fc_dbatch;
		int			fc_minbatch;
		size_t		fc_bucket_size;

		/*
		 * 如果结合hash_mem的并行哈希仍然需要多个批次，
		 * 我们将不得不回退到常规的hash_mem预算。
		 */
		if (fc_try_combined_hash_mem)
		{
			ExecChooseHashTableSize(fc_ntuples, fc_tupwidth, fc_useskew,
									false, fc_parallel_workers,
									fc_space_allowed,
									fc_numbuckets,
									fc_numbatches,
									fc_num_skew_mcvs);
			return;
		}

		/*
		 * 估计当hash_mem完全满时我们希望拥有的桶数量。 每个桶将包含一个桶指针加上
		 * NTUP_PER_BUCKET个元组，其预计大小已包含
		 * 哈希代码、指向下一个元组的指针等的开销。
		 */
		fc_bucket_size = (fc_tupsize * NTUP_PER_BUCKET + sizeof(HashJoinTuple));
		if (fc_hash_table_bytes <= fc_bucket_size)
			fc_sbuckets = 1;		/* 避免pg_nextpower2_size_t(0) */
		else
			fc_sbuckets = pg_nextpower2_size_t(fc_hash_table_bytes / fc_bucket_size);
		fc_sbuckets = Min(fc_sbuckets, fc_max_pointers);
		fc_nbuckets = (int) fc_sbuckets;
		fc_nbuckets = pg_nextpower2_32(fc_nbuckets);
		fc_bucket_bytes = fc_nbuckets * sizeof(HashJoinTuple);

		/*
		 * 桶是指向hashjoin元组的简单指针，而tupsize
		 * 包括指针、哈希代码和MinimalTupleData。 所以桶
		 * 实际上不应该超过hash_mem的25%（即使对于
		 * NTUP_PER_BUCKET=1）；也许对于不是
		 * 2^N字节的hash_mem值，我们可能会因为加倍而得到更多。所以让我们
		 * 在这里查找50%。
		 */
		Assert(fc_bucket_bytes <= fc_hash_table_bytes / 2);

		/* 计算所需的批次数。 */
		fc_dbatch = ceil(fc_inner_rel_bytes / (fc_hash_table_bytes - fc_bucket_bytes));
		fc_dbatch = Min(fc_dbatch, fc_max_pointers);
		fc_minbatch = (int) fc_dbatch;
		fc_nbatch = pg_nextpower2_32(Max(2, fc_minbatch));
	}

	Assert(fc_nbuckets > 0);
	Assert(fc_nbatch > 0);

	*fc_numbuckets = fc_nbuckets;
	*fc_numbatches = fc_nbatch;
}


/* ----------------------------------------------------------------
 *		ExecHashTableDestroy
 *
 *		销毁哈希表
 * ----------------------------------------------------------------
 */
void ExecHashTableDestroy(HashJoinTable hashtable)
{
	int			fc_i;

	/*
	 * 确保关闭所有临时文件。 我们跳过批次0，因为它
	 * 不能有任何临时文件（如果nbatch只有1，数组可能甚至不存在）。 并行哈希联接不使用这些文件。
	 */
	if (hashtable->innerBatchFile != NULL)
	{
		for (fc_i = 1; fc_i < hashtable->nbatch; fc_i++)
		{
			if (hashtable->innerBatchFile[fc_i])
				BufFileClose(hashtable->innerBatchFile[fc_i]);
			if (hashtable->outerBatchFile[fc_i])
				BufFileClose(hashtable->outerBatchFile[fc_i]);
		}
	}

	/* 释放工作内存（batchCxt是子项，因此它也将被销毁） */
	MemoryContextDelete(hashtable->hashCxt);

	/* 并丢弃控制块 */
	pfree(hashtable);
}

/*
 * ExecHashIncreaseNumBatches
 *		增加原始的批次数以减少
 *		当前的内存消耗
 */
static void fc_ExecHashIncreaseNumBatches(HashJoinTable hashtable)
{
	int			fc_oldnbatch = hashtable->nbatch;
	int			fc_curbatch = hashtable->curbatch;
	int			fc_nbatch;
	MemoryContext fc_oldcxt;
	long		fc_ninmemory;
	long		fc_nfreed;
	HashMemoryChunk fc_oldchunks;

	/* 如果我们决定停止增长，什么也不做 */
	if (!hashtable->growEnabled)
		return;

	/* 安全检查以避免溢出 */
	if (fc_oldnbatch > Min(INT_MAX / 2, MaxAllocSize / (sizeof(void *) * 2)))
		return;

	fc_nbatch = fc_oldnbatch * 2;
	Assert(fc_nbatch > 1);

#ifdef HJDEBUG
	printf("Hashjoin %p: increasing nbatch to %d because space = %zu\n",
		   hashtable, fc_nbatch, hashtable->spaceUsed);
#endif

	fc_oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);

	if (hashtable->innerBatchFile == NULL)
	{
		/* 我们之前没有文件数组 */
		hashtable->innerBatchFile = (BufFile **)
			palloc0(fc_nbatch * sizeof(BufFile *));
		hashtable->outerBatchFile = (BufFile **)
			palloc0(fc_nbatch * sizeof(BufFile *));
		/* 现在也是建立临时表空间的时候了 */
		PrepareTempTablespaces();
	}
	else
	{
		/* 扩大数组并将新增条目置为零 */
		hashtable->innerBatchFile = (BufFile **)
			repalloc(hashtable->innerBatchFile, fc_nbatch * sizeof(BufFile *));
		hashtable->outerBatchFile = (BufFile **)
			repalloc(hashtable->outerBatchFile, fc_nbatch * sizeof(BufFile *));
		MemSet(hashtable->innerBatchFile + fc_oldnbatch, 0,
			   (fc_nbatch - fc_oldnbatch) * sizeof(BufFile *));
		MemSet(hashtable->outerBatchFile + fc_oldnbatch, 0,
			   (fc_nbatch - fc_oldnbatch) * sizeof(BufFile *));
	}

	MemoryContextSwitchTo(fc_oldcxt);

	hashtable->nbatch = fc_nbatch;

	/*
	 * 扫描现有的哈希表条目，并转储出任何不再属于当前批次的条目。
	 */
	fc_ninmemory = fc_nfreed = 0;

	/* 如果知道我们需要调整nbuckets的大小，我们可以在重新批处理时进行调整。 */
	if (hashtable->nbuckets_optimal != hashtable->nbuckets)
	{
		/* 我们从不减少桶的数量 */
		Assert(hashtable->nbuckets_optimal > hashtable->nbuckets);

		hashtable->nbuckets = hashtable->nbuckets_optimal;
		hashtable->log2_nbuckets = hashtable->log2_nbuckets_optimal;

		hashtable->buckets.unshared =
			repalloc(hashtable->buckets.unshared,
					 sizeof(HashJoinTuple) * hashtable->nbuckets);
	}

	/*
	 * 我们将直接扫描这些块，以便现在重置桶，而不必跟踪哪个元组在桶中
	 * 已经被处理过。我们会在进行时释放旧的块。
	 */
	memset(hashtable->buckets.unshared, 0,
		   sizeof(HashJoinTuple) * hashtable->nbuckets);
	fc_oldchunks = hashtable->chunks;
	hashtable->chunks = NULL;

	/* 那么，让我们扫描旧块中的所有元组 */
	while (fc_oldchunks != NULL)
	{
		HashMemoryChunk fc_nextchunk = fc_oldchunks->next.unshared;

		/* 在缓冲区中的位置（最多到oldchunks->used） */
		size_t		fc_idx = 0;

		/* 处理存储在此块中的所有元组（然后释放它） */
		while (fc_idx < fc_oldchunks->used)
		{
			HashJoinTuple fc_hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(fc_oldchunks) + fc_idx);
			MinimalTuple fc_tuple = HJTUPLE_MINTUPLE(fc_hashTuple);
			int			fc_hashTupleSize = (HJTUPLE_OVERHEAD + fc_tuple->t_len);
			int			fc_bucketno;
			int			fc_batchno;

			fc_ninmemory++;
			ExecHashGetBucketAndBatch(hashtable, fc_hashTuple->hashvalue,
									  &fc_bucketno, &fc_batchno);

			if (fc_batchno == fc_curbatch)
			{
				/* 将元组保留在内存中 - 复制到新的块中 */
				HashJoinTuple fc_copyTuple;

				fc_copyTuple = (HashJoinTuple) fc_dense_alloc(hashtable, fc_hashTupleSize);
				memcpy(fc_copyTuple, fc_hashTuple, fc_hashTupleSize);

				/* 并将其添加回适当的桶中 */
				fc_copyTuple->next.unshared = hashtable->buckets.unshared[fc_bucketno];
				hashtable->buckets.unshared[fc_bucketno] = fc_copyTuple;
			}
			else
			{
				/* 将其转储 */
				Assert(fc_batchno > fc_curbatch);
				ExecHashJoinSaveTuple(HJTUPLE_MINTUPLE(fc_hashTuple),
									  fc_hashTuple->hashvalue,
									  &hashtable->innerBatchFile[fc_batchno]);

				hashtable->spaceUsed -= fc_hashTupleSize;
				fc_nfreed++;
			}

			/* 此块中的下一个元组 */
			fc_idx += MAXALIGN(fc_hashTupleSize);

			/* 允许此循环可取消 */
			CHECK_FOR_INTERRUPTS();
		}

		/* 我们完成了这个块 - 释放它并继续下一个块 */
		pfree(fc_oldchunks);
		fc_oldchunks = fc_nextchunk;
	}

#ifdef HJDEBUG
	printf("Hashjoin %p: freed %ld of %ld tuples, space now %zu\n",
		   hashtable, fc_nfreed, fc_ninmemory, hashtable->spaceUsed);
#endif

	/*
	 * 如果我们转储了表中的所有或没有元组，禁用
	 * nbatch的进一步扩展。 这种情况意味着我们有
	 * 足够的相同hashvalues的元组以溢出spaceAllowed。
	 * 增加nbatch是无法解决的，因为没有办法进一步细分
	 * 这个组。我们必须忍受并希望服务器有足够的内存。
	 */
	if (fc_nfreed == 0 || fc_nfreed == fc_ninmemory)
	{
		hashtable->growEnabled = false;
#ifdef HJDEBUG
		printf("Hashjoin %p: disabling further increase of nbatch\n",
			   hashtable);
#endif
	}
}

/*
 * ExecParallelHashIncreaseNumBatches
 *		每个附加到grow_batches_barrier的参与者必须在观察到
 *		growth == PHJ_GROWTH_NEED_MORE_BATCHES时运行此
 *		函数。
 */
static void fc_ExecParallelHashIncreaseNumBatches(HashJoinTable hashtable)
{
	ParallelHashJoinState *fc_pstate = hashtable->parallel_state;
	int			fc_i;

	Assert(BarrierPhase(&fc_pstate->build_barrier) == PHJ_BUILD_HASHING_INNER);

	/*
	 * 这不太可能，但我们需要为新参与者在我们进行这项操作时出现做好准备，所以我们需要在这里打开
	 * barrier阶段。
	 */
	switch (PHJ_GROW_BATCHES_PHASE(BarrierPhase(&fc_pstate->grow_batches_barrier)))
	{
		case PHJ_GROW_BATCHES_ELECTING:

			/*
			 * 选举一个参与者准备增加批次数。
			 * 这涉及重新分配或重置批次0的桶
			 * 为所有参与者开始重新分区元组做准备。
			 */
			if (BarrierArriveAndWait(&fc_pstate->grow_batches_barrier,
									 WAIT_EVENT_HASH_GROW_BATCHES_ELECT))
			{
				dsa_pointer_atomic *fc_buckets;
				ParallelHashJoinBatch *fc_old_batch0;
				int			fc_new_nbatch;
				int			fc_i;

				/* 将旧批次移开。 */
				fc_old_batch0 = hashtable->batches[0].shared;
				fc_pstate->old_batches = fc_pstate->batches;
				fc_pstate->old_nbatch = hashtable->nbatch;
				fc_pstate->batches = InvalidDsaPointer;

				/* 释放该后端的旧访问器。 */
				fc_ExecParallelHashCloseBatchAccessors(hashtable);

				/* 确定使用多少批次。 */
				if (hashtable->nbatch == 1)
				{
					/*
					 * 我们正在从单批次转变为多批次。我们需要
					 * 从一个大型组合内存预算切换到
					 * 常规hash_mem预算。
					 */
					fc_pstate->space_allowed = get_hash_memory_limit();

					/*
					 * 所有参与者的组合hash_mem不足。
					 * 因此每个参与者一个批次大致相当，并且可能也
					 * 不足。因此尝试每个参与者两个批次，
					 * 并向上取整到二的幂。
					 */
					fc_new_nbatch = pg_nextpower2_32(fc_pstate->nparticipants * 2);
				}
				else
				{
					/*
					 * 我们已经处于多批次状态。尝试将
					 * 批次数加倍。
					 */
					fc_new_nbatch = hashtable->nbatch * 2;
				}

				/* 分配新的更大批次代。 */
				Assert(hashtable->nbatch == fc_pstate->nbatch);
				fc_ExecParallelHashJoinSetUpBatches(hashtable, fc_new_nbatch);
				Assert(hashtable->nbatch == fc_pstate->nbatch);

				/* 替换或回收批次 0 的桶数组。 */
				if (fc_pstate->old_nbatch == 1)
				{
					double		fc_dtuples;
					double		fc_dbuckets;
					int			fc_new_nbuckets;
					uint32		fc_max_buckets;

					/*
					 * 我们可能还需要一个更小的桶数组。每个批次我们预期有多少个元组，假设我们到目前为止只有一半？通常我们不需要更改桶数组的大小，因为随着我们添加更多的批次，每个批次的大小保持不变，但在这种特殊情况下，我们从一个大批次变为多个小批次，继续保留大数组将是浪费。
					 */
					fc_dtuples = (fc_old_batch0->ntuples * 2.0) / fc_new_nbatch;

					/*
					 * 我们需要计算最大桶数，以保持在 MaxAllocSize 边界内。向前一位2的幂取整最大数量，因为稍后我们将该数字取整到下一个2的幂。
					 */
					fc_max_buckets = pg_prevpower2_32((uint32)
												   (MaxAllocSize / sizeof(dsa_pointer_atomic)));
					fc_dbuckets = ceil(fc_dtuples / NTUP_PER_BUCKET);
					fc_dbuckets = Min(fc_dbuckets, fc_max_buckets);
					fc_new_nbuckets = (int) fc_dbuckets;
					fc_new_nbuckets = Max(fc_new_nbuckets, 1024);
					fc_new_nbuckets = pg_nextpower2_32(fc_new_nbuckets);
					dsa_free(hashtable->area, fc_old_batch0->buckets);
					hashtable->batches[0].shared->buckets =
						dsa_allocate(hashtable->area,
									 sizeof(dsa_pointer_atomic) * fc_new_nbuckets);
					fc_buckets = (dsa_pointer_atomic *)
						dsa_get_address(hashtable->area,
										hashtable->batches[0].shared->buckets);
					for (fc_i = 0; fc_i < fc_new_nbuckets; ++fc_i)
						dsa_pointer_atomic_init(&fc_buckets[fc_i], InvalidDsaPointer);
					fc_pstate->nbuckets = fc_new_nbuckets;
				}
				else
				{
					/* 回收现有的桶数组。 */
					hashtable->batches[0].shared->buckets = fc_old_batch0->buckets;
					fc_buckets = (dsa_pointer_atomic *)
						dsa_get_address(hashtable->area, fc_old_batch0->buckets);
					for (fc_i = 0; fc_i < hashtable->nbuckets; ++fc_i)
						dsa_pointer_atomic_write(&fc_buckets[fc_i], InvalidDsaPointer);
				}

				/* 将所有块移动到工作队列以进行并行处理。 */
				fc_pstate->chunk_work_queue = fc_old_batch0->chunks;

				/* 在我们增长的时候暂时禁用进一步增长。 */
				fc_pstate->growth = PHJ_GROWTH_DISABLED;
			}
			else
			{
				/* 所有其他参与者只需将他们的元组刷新到磁盘。 */
				fc_ExecParallelHashCloseBatchAccessors(hashtable);
			}
			/* Fall through. */

		case PHJ_GROW_BATCHES_ALLOCATING:
			/* 等待以上操作完成。 */
			BarrierArriveAndWait(&fc_pstate->grow_batches_barrier,
								 WAIT_EVENT_HASH_GROW_BATCHES_ALLOCATE);
			/* Fall through. */

		case PHJ_GROW_BATCHES_REPARTITIONING:
			/* 确保我们拥有当前的维度和桶。 */
			fc_ExecParallelHashEnsureBatchAccessors(hashtable);
			ExecParallelHashTableSetCurrentBatch(hashtable, 0);
			/* 然后进行分区，刷新计数器。 */
			fc_ExecParallelHashRepartitionFirst(hashtable);
			fc_ExecParallelHashRepartitionRest(hashtable);
			fc_ExecParallelHashMergeCounters(hashtable);
			/* 等待以上操作完成。 */
			BarrierArriveAndWait(&fc_pstate->grow_batches_barrier,
								 WAIT_EVENT_HASH_GROW_BATCHES_REPARTITION);
			/* Fall through. */

		case PHJ_GROW_BATCHES_DECIDING:

			/*
			 * 选择一个参与者进行清理，并决定是否需要进一步的重新分区，或是否应禁用，因为这没有帮助。
			 */
			if (BarrierArriveAndWait(&fc_pstate->grow_batches_barrier,
									 WAIT_EVENT_HASH_GROW_BATCHES_DECIDE))
			{
				ParallelHashJoinBatch *fc_old_batches;
				bool		fc_space_exhausted = false;
				bool		fc_extreme_skew_detected = false;

				/* 确保我们拥有当前的维度和桶。 */
				fc_ExecParallelHashEnsureBatchAccessors(hashtable);
				ExecParallelHashTableSetCurrentBatch(hashtable, 0);

				fc_old_batches = dsa_get_address(hashtable->area, fc_pstate->old_batches);

				/* 新一代批次是否有疲惫的？ */
				for (fc_i = 0; fc_i < hashtable->nbatch; ++fc_i)
				{
					ParallelHashJoinBatch *fc_batch;
					ParallelHashJoinBatch *fc_old_batch;
					int			fc_parent;

					fc_batch = hashtable->batches[fc_i].shared;
					if (fc_batch->space_exhausted ||
						fc_batch->estimated_size > fc_pstate->space_allowed)
						fc_space_exhausted = true;

					fc_parent = fc_i % fc_pstate->old_nbatch;
					fc_old_batch = NthParallelHashJoinBatch(fc_old_batches, fc_parent);
					if (fc_old_batch->space_exhausted ||
						fc_batch->estimated_size > fc_pstate->space_allowed)
					{
						/*
						 * 这个批次是否接收到来自其父批次的所有元组？这表明进一步的重新分区不会有帮助（哈希值可能都一样）。
						 */
						if (fc_batch->ntuples == hashtable->batches[fc_parent].shared->old_ntuples)
							fc_extreme_skew_detected = true;
					}
				}

				/* 如果没有帮助或我们会溢出，就不要继续增长。 */
				if (fc_extreme_skew_detected || hashtable->nbatch >= INT_MAX / 2)
					fc_pstate->growth = PHJ_GROWTH_DISABLED;
				else if (fc_space_exhausted)
					fc_pstate->growth = PHJ_GROWTH_NEED_MORE_BATCHES;
				else
					fc_pstate->growth = PHJ_GROWTH_OK;

				/* 释放共享内存中的旧批次。 */
				dsa_free(hashtable->area, fc_pstate->old_batches);
				fc_pstate->old_batches = InvalidDsaPointer;
			}
			/* Fall through. */

		case PHJ_GROW_BATCHES_FINISHING:
			/* 等待以上操作完成。 */
			BarrierArriveAndWait(&fc_pstate->grow_batches_barrier,
								 WAIT_EVENT_HASH_GROW_BATCHES_FINISH);
	}
}

/*
 * 重新分区当前加载到内存中的元组以进行内部批次 0，因为批次数量已增加。一些元组保留在内存中，一些写入到后面的批次。
 */
static void fc_ExecParallelHashRepartitionFirst(HashJoinTable hashtable)
{
	dsa_pointer fc_chunk_shared;
	HashMemoryChunk fc_chunk;

	Assert(hashtable->nbatch == hashtable->parallel_state->nbatch);

	while ((fc_chunk = fc_ExecParallelHashPopChunkQueue(hashtable, &fc_chunk_shared)))
	{
		size_t		fc_idx = 0;

		/* 重新分区这个块中的所有元组。 */
		while (fc_idx < fc_chunk->used)
		{
			HashJoinTuple fc_hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(fc_chunk) + fc_idx);
			MinimalTuple fc_tuple = HJTUPLE_MINTUPLE(fc_hashTuple);
			HashJoinTuple fc_copyTuple;
			dsa_pointer fc_shared;
			int			fc_bucketno;
			int			fc_batchno;

			ExecHashGetBucketAndBatch(hashtable, fc_hashTuple->hashvalue,
									  &fc_bucketno, &fc_batchno);

			Assert(fc_batchno < hashtable->nbatch);
			if (fc_batchno == 0)
			{
				/* 它仍然属于批次 0。复制到一个新块。 */
				fc_copyTuple =
					fc_ExecParallelHashTupleAlloc(hashtable,
											   HJTUPLE_OVERHEAD + fc_tuple->t_len,
											   &fc_shared);
				fc_copyTuple->hashvalue = fc_hashTuple->hashvalue;
				memcpy(HJTUPLE_MINTUPLE(fc_copyTuple), fc_tuple, fc_tuple->t_len);
				fc_ExecParallelHashPushTuple(&hashtable->buckets.shared[fc_bucketno],
										  fc_copyTuple, fc_shared);
			}
			else
			{
				size_t		fc_tuple_size =
				MAXALIGN(HJTUPLE_OVERHEAD + fc_tuple->t_len);

				/* 它属于后面的批次。 */
				hashtable->batches[fc_batchno].estimated_size += fc_tuple_size;
				sts_puttuple(hashtable->batches[fc_batchno].inner_tuples,
							 &fc_hashTuple->hashvalue, fc_tuple);
			}

			/* 计数字元组。 */
			++hashtable->batches[0].old_ntuples;
			++hashtable->batches[fc_batchno].ntuples;

			fc_idx += MAXALIGN(HJTUPLE_OVERHEAD +
							HJTUPLE_MINTUPLE(fc_hashTuple)->t_len);
		}

		/* 释放这个块。 */
		dsa_free(hashtable->area, fc_chunk_shared);

		CHECK_FOR_INTERRUPTS();
	}
}

/*
 * 帮助重新分区内部批次 1..n。
 */
static void fc_ExecParallelHashRepartitionRest(HashJoinTable hashtable)
{
	ParallelHashJoinState *fc_pstate = hashtable->parallel_state;
	int			fc_old_nbatch = fc_pstate->old_nbatch;
	SharedTuplestoreAccessor **fc_old_inner_tuples;
	ParallelHashJoinBatch *fc_old_batches;
	int			fc_i;

	/* 获取我们手上的上一代批次。 */
	fc_old_batches = (ParallelHashJoinBatch *)
		dsa_get_address(hashtable->area, fc_pstate->old_batches);
	fc_old_inner_tuples = palloc0(sizeof(SharedTuplestoreAccessor *) * fc_old_nbatch);
	for (fc_i = 1; fc_i < fc_old_nbatch; ++fc_i)
	{
		ParallelHashJoinBatch *fc_shared =
		NthParallelHashJoinBatch(fc_old_batches, fc_i);

		fc_old_inner_tuples[fc_i] = sts_attach(ParallelHashJoinBatchInner(fc_shared),
										 ParallelWorkerNumber + 1,
										 &fc_pstate->fileset);
	}

	/* 加入重新分区的努力。 */
	for (fc_i = 1; fc_i < fc_old_nbatch; ++fc_i)
	{
		MinimalTuple fc_tuple;
		uint32		fc_hashvalue;

		/* 扫描上一代的一个分区。 */
		sts_begin_parallel_scan(fc_old_inner_tuples[fc_i]);
		while ((fc_tuple = sts_parallel_scan_next(fc_old_inner_tuples[fc_i], &fc_hashvalue)))
		{
			size_t		fc_tuple_size = MAXALIGN(HJTUPLE_OVERHEAD + fc_tuple->t_len);
			int			fc_bucketno;
			int			fc_batchno;

			/* 决定它在新一代中的分区。 */
			ExecHashGetBucketAndBatch(hashtable, fc_hashvalue, &fc_bucketno,
									  &fc_batchno);

			hashtable->batches[fc_batchno].estimated_size += fc_tuple_size;
			++hashtable->batches[fc_batchno].ntuples;
			++hashtable->batches[fc_i].old_ntuples;

			/* 存储元组到其新的批次。 */
			sts_puttuple(hashtable->batches[fc_batchno].inner_tuples,
						 &fc_hashvalue, fc_tuple);

			CHECK_FOR_INTERRUPTS();
		}
		sts_end_parallel_scan(fc_old_inner_tuples[fc_i]);
	}

	pfree(fc_old_inner_tuples);
}

/*
 * 将后端本地的每个批次计数器转移到共享总数中。
 */
static void fc_ExecParallelHashMergeCounters(HashJoinTable hashtable)
{
	ParallelHashJoinState *fc_pstate = hashtable->parallel_state;
	int			fc_i;

	LWLockAcquire(&fc_pstate->lock, LW_EXCLUSIVE);
	fc_pstate->total_tuples = 0;
	for (fc_i = 0; fc_i < hashtable->nbatch; ++fc_i)
	{
		ParallelHashJoinBatchAccessor *fc_batch = &hashtable->batches[fc_i];

		fc_batch->shared->size += fc_batch->size;
		fc_batch->shared->estimated_size += fc_batch->estimated_size;
		fc_batch->shared->ntuples += fc_batch->ntuples;
		fc_batch->shared->old_ntuples += fc_batch->old_ntuples;
		fc_batch->size = 0;
		fc_batch->estimated_size = 0;
		fc_batch->ntuples = 0;
		fc_batch->old_ntuples = 0;
		fc_pstate->total_tuples += fc_batch->shared->ntuples;
	}
	LWLockRelease(&fc_pstate->lock);
}

/*
 * ExecHashIncreaseNumBuckets
 *		增加原始的桶数以减少每个桶中的元组数
 */
static void fc_ExecHashIncreaseNumBuckets(HashJoinTable hashtable)
{
	HashMemoryChunk fc_chunk;

	/* 如果没有增加则不执行任何操作（之所以叫增加是有原因的） */
	if (hashtable->nbuckets >= hashtable->nbuckets_optimal)
		return;

#ifdef HJDEBUG
	printf("Hashjoin %p: increasing nbuckets %d => %d\n",
		   hashtable, hashtable->nbuckets, hashtable->nbuckets_optimal);
#endif

	hashtable->nbuckets = hashtable->nbuckets_optimal;
	hashtable->log2_nbuckets = hashtable->log2_nbuckets_optimal;

	Assert(hashtable->nbuckets > 1);
	Assert(hashtable->nbuckets <= (INT_MAX / 2));
	Assert(hashtable->nbuckets == (1 << hashtable->log2_nbuckets));

	/*
	 * 只需重新分配适当数量的桶 - 我们不需要遍历它们 -
	 * 我们可以遍历密集分配的块（就像在 ExecHashIncreaseNumBatches 中，
	 * 但不需要将所有内容复制到新的块中）
	 */
	hashtable->buckets.unshared =
		(HashJoinTuple *) repalloc(hashtable->buckets.unshared,
								   hashtable->nbuckets * sizeof(HashJoinTuple));

	memset(hashtable->buckets.unshared, 0,
		   hashtable->nbuckets * sizeof(HashJoinTuple));

	/* 扫描所有块中的所有元组以重建哈希表 */
	for (fc_chunk = hashtable->chunks; fc_chunk != NULL; fc_chunk = fc_chunk->next.unshared)
	{
		/* 处理存储在此块中的所有元组 */
		size_t		fc_idx = 0;

		while (fc_idx < fc_chunk->used)
		{
			HashJoinTuple fc_hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(fc_chunk) + fc_idx);
			int			fc_bucketno;
			int			fc_batchno;

			ExecHashGetBucketAndBatch(hashtable, fc_hashTuple->hashvalue,
									  &fc_bucketno, &fc_batchno);

			/* 将元组添加到适当的桶中 */
			fc_hashTuple->next.unshared = hashtable->buckets.unshared[fc_bucketno];
			hashtable->buckets.unshared[fc_bucketno] = fc_hashTuple;

			/* 将索引推进到元组后面 */
			fc_idx += MAXALIGN(HJTUPLE_OVERHEAD +
							HJTUPLE_MINTUPLE(fc_hashTuple)->t_len);
		}

		/* 允许此循环可取消 */
		CHECK_FOR_INTERRUPTS();
	}
}

static void fc_ExecParallelHashIncreaseNumBuckets(HashJoinTable hashtable)
{
	ParallelHashJoinState *fc_pstate = hashtable->parallel_state;
	int			fc_i;
	HashMemoryChunk fc_chunk;
	dsa_pointer fc_chunk_s;

	Assert(BarrierPhase(&fc_pstate->build_barrier) == PHJ_BUILD_HASHING_INNER);

	/*
	 * 这不太可能，但我们需要为新参与者在我们进行这项操作时出现做好准备，所以我们需要在这里打开
	 * barrier阶段。
	 */
	switch (PHJ_GROW_BUCKETS_PHASE(BarrierPhase(&fc_pstate->grow_buckets_barrier)))
	{
		case PHJ_GROW_BUCKETS_ELECTING:
			/* 选举一个参与者准备增加 nbuckets。 */
			if (BarrierArriveAndWait(&fc_pstate->grow_buckets_barrier,
									 WAIT_EVENT_HASH_GROW_BUCKETS_ELECT))
			{
				size_t		fc_size;
				dsa_pointer_atomic *fc_buckets;

				/* 将桶数组的大小加倍。 */
				fc_pstate->nbuckets *= 2;
				fc_size = fc_pstate->nbuckets * sizeof(dsa_pointer_atomic);
				hashtable->batches[0].shared->size += fc_size / 2;
				dsa_free(hashtable->area, hashtable->batches[0].shared->buckets);
				hashtable->batches[0].shared->buckets =
					dsa_allocate(hashtable->area, fc_size);
				fc_buckets = (dsa_pointer_atomic *)
					dsa_get_address(hashtable->area,
									hashtable->batches[0].shared->buckets);
				for (fc_i = 0; fc_i < fc_pstate->nbuckets; ++fc_i)
					dsa_pointer_atomic_init(&fc_buckets[fc_i], InvalidDsaPointer);

				/* 将块列表放入工作队列中。 */
				fc_pstate->chunk_work_queue = hashtable->batches[0].shared->chunks;

				/* 清除标志。 */
				fc_pstate->growth = PHJ_GROWTH_OK;
			}
			/* Fall through. */

		case PHJ_GROW_BUCKETS_ALLOCATING:
			/* 等待以上操作完成。 */
			BarrierArriveAndWait(&fc_pstate->grow_buckets_barrier,
								 WAIT_EVENT_HASH_GROW_BUCKETS_ALLOCATE);
			/* Fall through. */

		case PHJ_GROW_BUCKETS_REINSERTING:
			/* 将所有元组重新插入到哈希表中。 */
			fc_ExecParallelHashEnsureBatchAccessors(hashtable);
			ExecParallelHashTableSetCurrentBatch(hashtable, 0);
			while ((fc_chunk = fc_ExecParallelHashPopChunkQueue(hashtable, &fc_chunk_s)))
			{
				size_t		fc_idx = 0;

				while (fc_idx < fc_chunk->used)
				{
					HashJoinTuple fc_hashTuple = (HashJoinTuple) (HASH_CHUNK_DATA(fc_chunk) + fc_idx);
					dsa_pointer fc_shared = fc_chunk_s + HASH_CHUNK_HEADER_SIZE + fc_idx;
					int			fc_bucketno;
					int			fc_batchno;

					ExecHashGetBucketAndBatch(hashtable, fc_hashTuple->hashvalue,
											  &fc_bucketno, &fc_batchno);
					Assert(fc_batchno == 0);

					/* 将元组添加到适当的桶中 */
					fc_ExecParallelHashPushTuple(&hashtable->buckets.shared[fc_bucketno],
											  fc_hashTuple, fc_shared);

					/* 将索引推进到元组后面 */
					fc_idx += MAXALIGN(HJTUPLE_OVERHEAD +
									HJTUPLE_MINTUPLE(fc_hashTuple)->t_len);
				}

				/* 允许此循环可取消 */
				CHECK_FOR_INTERRUPTS();
			}
			BarrierArriveAndWait(&fc_pstate->grow_buckets_barrier,
								 WAIT_EVENT_HASH_GROW_BUCKETS_REINSERT);
	}
}

/*
 * ExecHashTableInsert
 *	根据哈希值将元组插入哈希表
 *	它可能仅仅会进入一个临时文件以供稍后的批处理使用
 *
 * 注意：传入的 TupleTableSlot 可能包含一般的、最小的或虚拟的
 * 元组；在从批处理文件重新加载元组时，特别是最小的情况
 * 是必然会发生的。我们在一般元组的情况下可以节省一些周期，
 * 通过不强制将槽内容转换为最小形式；是否值得为了所需的混乱而
 * 这样做尚不明确。
 */
void ExecHashTableInsert(HashJoinTable hashtable,
					TupleTableSlot *fc_slot,
					uint32 fc_hashvalue)
{
	bool		fc_shouldFree;
	MinimalTuple fc_tuple = ExecFetchSlotMinimalTuple(fc_slot, &fc_shouldFree);
	int			fc_bucketno;
	int			fc_batchno;

	ExecHashGetBucketAndBatch(hashtable, fc_hashvalue,
							  &fc_bucketno, &fc_batchno);

	/*
	 * 决定将元组放入哈希表还是临时文件
	 */
	if (fc_batchno == hashtable->curbatch)
	{
		/*
		 * 将元组放入哈希表
		 */
		HashJoinTuple fc_hashTuple;
		int			fc_hashTupleSize;
		double		fc_ntuples = (hashtable->totalTuples - hashtable->skewTuples);

		/* 创建 HashJoinTuple */
		fc_hashTupleSize = HJTUPLE_OVERHEAD + fc_tuple->t_len;
		fc_hashTuple = (HashJoinTuple) fc_dense_alloc(hashtable, fc_hashTupleSize);

		fc_hashTuple->hashvalue = fc_hashvalue;
		memcpy(HJTUPLE_MINTUPLE(fc_hashTuple), fc_tuple, fc_tuple->t_len);

		/*
		 * 我们在插入时总是重置元组匹配标志。这是可以的，
		 * 即使在从批处理文件重新加载元组时，因为该元组
		 * 在进入批处理文件之前不可能已经与外部元组匹配。
		 */
		HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(fc_hashTuple));

		/* 将其推到桶列表的前面 */
		fc_hashTuple->next.unshared = hashtable->buckets.unshared[fc_bucketno];
		hashtable->buckets.unshared[fc_bucketno] = fc_hashTuple;

		/*
		 * 如果我们刚刚超过了 NTUP_PER_BUCKET 阈值，则增加
		 * （最佳）桶的数量，但仅在仍然有单个批处理时。
		 */
		if (hashtable->nbatch == 1 &&
			fc_ntuples > (hashtable->nbuckets_optimal * NTUP_PER_BUCKET))
		{
			/* 防止整数溢出和分配大小溢出 */
			if (hashtable->nbuckets_optimal <= INT_MAX / 2 &&
				hashtable->nbuckets_optimal * 2 <= MaxAllocSize / sizeof(HashJoinTuple))
			{
				hashtable->nbuckets_optimal *= 2;
				hashtable->log2_nbuckets_optimal += 1;
			}
		}

		/* 计算使用的空间，如果使用过多则退回 */
		hashtable->spaceUsed += fc_hashTupleSize;
		if (hashtable->spaceUsed > hashtable->spacePeak)
			hashtable->spacePeak = hashtable->spaceUsed;
		if (hashtable->spaceUsed +
			hashtable->nbuckets_optimal * sizeof(HashJoinTuple)
			> hashtable->spaceAllowed)
			fc_ExecHashIncreaseNumBatches(hashtable);
	}
	else
	{
		/*
		 * 将元组放入临时文件以供稍后的批处理使用
		 */
		Assert(fc_batchno > hashtable->curbatch);
		ExecHashJoinSaveTuple(fc_tuple,
							  fc_hashvalue,
							  &hashtable->innerBatchFile[fc_batchno]);
	}

	if (fc_shouldFree)
		heap_free_minimal_tuple(fc_tuple);
}

/*
 * ExecParallelHashTableInsert
 *	将元组插入到共享哈希表或共享批处理元组存储中
 */
void ExecParallelHashTableInsert(HashJoinTable hashtable,
							TupleTableSlot *fc_slot,
							uint32 fc_hashvalue)
{
	bool		fc_shouldFree;
	MinimalTuple fc_tuple = ExecFetchSlotMinimalTuple(fc_slot, &fc_shouldFree);
	dsa_pointer fc_shared;
	int			fc_bucketno;
	int			fc_batchno;

retry:
	ExecHashGetBucketAndBatch(hashtable, fc_hashvalue, &fc_bucketno, &fc_batchno);

	if (fc_batchno == 0)
	{
		HashJoinTuple fc_hashTuple;

		/* 尝试将其加载到内存中。 */
		Assert(BarrierPhase(&hashtable->parallel_state->build_barrier) ==
			   PHJ_BUILD_HASHING_INNER);
		fc_hashTuple = fc_ExecParallelHashTupleAlloc(hashtable,
											   HJTUPLE_OVERHEAD + fc_tuple->t_len,
											   &fc_shared);
		if (fc_hashTuple == NULL)
			goto retry;

		/* 将哈希值存储在 HashJoinTuple 头中。 */
		fc_hashTuple->hashvalue = fc_hashvalue;
		memcpy(HJTUPLE_MINTUPLE(fc_hashTuple), fc_tuple, fc_tuple->t_len);

		/* 将其推到桶列表的前面 */
		fc_ExecParallelHashPushTuple(&hashtable->buckets.shared[fc_bucketno],
								  fc_hashTuple, fc_shared);
	}
	else
	{
		size_t		fc_tuple_size = MAXALIGN(HJTUPLE_OVERHEAD + fc_tuple->t_len);

		Assert(fc_batchno > 0);

		/* 如果需要，尝试在批处理中预分配空间。 */
		if (hashtable->batches[fc_batchno].preallocated < fc_tuple_size)
		{
			if (!fc_ExecParallelHashTuplePrealloc(hashtable, fc_batchno, fc_tuple_size))
				goto retry;
		}

		Assert(hashtable->batches[fc_batchno].preallocated >= fc_tuple_size);
		hashtable->batches[fc_batchno].preallocated -= fc_tuple_size;
		sts_puttuple(hashtable->batches[fc_batchno].inner_tuples, &fc_hashvalue,
					 fc_tuple);
	}
	++hashtable->batches[fc_batchno].ntuples;

	if (fc_shouldFree)
		heap_free_minimal_tuple(fc_tuple);
}

/*
 * 将元组插入当前哈希表。与 ExecParallelHashTableInsert 不同，
 * 此版本不准备将元组发送到其他批处理或一旦在内存不足的情况下
 * 运行，应仅在禁用增长后与属于当前批处理的元组一起调用。
 */
void ExecParallelHashTableInsertCurrentBatch(HashJoinTable hashtable,
										TupleTableSlot *fc_slot,
										uint32 fc_hashvalue)
{
	bool		fc_shouldFree;
	MinimalTuple fc_tuple = ExecFetchSlotMinimalTuple(fc_slot, &fc_shouldFree);
	HashJoinTuple fc_hashTuple;
	dsa_pointer fc_shared;
	int			fc_batchno;
	int			fc_bucketno;

	ExecHashGetBucketAndBatch(hashtable, fc_hashvalue, &fc_bucketno, &fc_batchno);
	Assert(fc_batchno == hashtable->curbatch);
	fc_hashTuple = fc_ExecParallelHashTupleAlloc(hashtable,
										   HJTUPLE_OVERHEAD + fc_tuple->t_len,
										   &fc_shared);
	fc_hashTuple->hashvalue = fc_hashvalue;
	memcpy(HJTUPLE_MINTUPLE(fc_hashTuple), fc_tuple, fc_tuple->t_len);
	HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(fc_hashTuple));
	fc_ExecParallelHashPushTuple(&hashtable->buckets.shared[fc_bucketno],
							  fc_hashTuple, fc_shared);

	if (fc_shouldFree)
		heap_free_minimal_tuple(fc_tuple);
}

/*
 * ExecHashGetHashValue
 *		计算元组的哈希值
 *
 * 要测试的元组必须在 econtext->ecxt_outertuple 中（因此哈希键表达式中的
 * Vars 需要具有 OUTER_VAR 作为 varno）。如果 outer_tuple 为假
 * （意味着它是 HashJoin 的内部节点 Hash），则 econtext、
 * hashkeys 和槽需要来自 Hash，hashkeys/slot 需要引用并适合来自
 * Hash 下面节点的元组。相反，如果 outer_tuple 为真，则 econtext
 * 来自 HashJoin，并且 hashkeys/slot 需要适合来自 HashJoin 的外部节点的元组。
 *
 * 结果为真的意味着元组的哈希值已成功计算并存储在 *hashvalue 中。
 * 结果为假意味着元组不能匹配，因为它包含一个空属性，因此
 * 应立即丢弃它。（如果 keep_nulls 为真，则永远不会返回假。）
 */
bool ExecHashGetHashValue(HashJoinTable hashtable,
					 ExprContext *fc_econtext,
					 List *fc_hashkeys,
					 bool fc_outer_tuple,
					 bool fc_keep_nulls,
					 uint32 *fc_hashvalue)
{
	uint32		fc_hashkey = 0;
	FmgrInfo   *fc_hashfunctions;
	ListCell   *fc_hk;
	int			fc_i = 0;
	MemoryContext fc_oldContext;

	/*
	 * 我们每次重置评估上下文以回收在哈希键表达式中泄漏的任何内存。
	 */
	ResetExprContext(fc_econtext);

	fc_oldContext = MemoryContextSwitchTo(fc_econtext->ecxt_per_tuple_memory);

	if (fc_outer_tuple)
		fc_hashfunctions = hashtable->outer_hashfunctions;
	else
		fc_hashfunctions = hashtable->inner_hashfunctions;

	foreach(fc_hk, fc_hashkeys)
	{
		ExprState  *fc_keyexpr = (ExprState *) lfirst(fc_hk);
		Datum		fc_keyval;
		bool		fc_isNull;

		/* 通过旋转组合连续的哈希键 */
		fc_hashkey = pg_rotate_left32(fc_hashkey, 1);

		/*
		 * 获取元组的连接属性值
		 */
		fc_keyval = ExecEvalExpr(fc_keyexpr, fc_econtext, &fc_isNull);

		/*
		 * 如果属性为 NULL，并且连接操作符是严格的，那么这个元组不能通过连接条件，
		 * 因此我们可以立即将其拒绝（除非我们正在扫描外连接的外部，
		 * 在这种情况下我们不能拒绝它）。否则，我们假设 NULL 的哈希码为零
		 * （这将支持像 IS NOT DISTINCT 这样的操作符，但不会出现更随机的行为）。
		 * 我们将哈希支持函数视为严格的，即使操作符不是。
		 *
		 * 注意：目前，所有可哈希连接的操作符都必须是严格的，因为哈希索引 AM 假定这样。
		 * 但是，在这里允许非严格只需要很少的额外代码，因此我们可以这样做。
		 */
		if (fc_isNull)
		{
			if (hashtable->hashStrict[fc_i] && !fc_keep_nulls)
			{
				MemoryContextSwitchTo(fc_oldContext);
				return false;	/*无法匹配* */
			}
			/*否则，保持hashkey不变，相当于hashcode 0* */
		}
		else
		{
			/*计算哈希函数* */
			uint32		fc_hkey;

			fc_hkey = DatumGetUInt32(FunctionCall1Coll(&fc_hashfunctions[fc_i], hashtable->collations[fc_i], fc_keyval));
			fc_hashkey ^= fc_hkey;
		}

		fc_i++;
	}

	MemoryContextSwitchTo(fc_oldContext);

	*fc_hashvalue = fc_hashkey;
	return true;
}

/* 
 * ExecHashGetBucketAndBatch
 *		确定哈希值的桶号和批次号
 *
 * 注意：动态增加nbatch必须不会改变给定哈希代码的桶号 
 * （因为我们不将元组移动到不同的哈希链），并且只能使批次号保持相同或增加。我们的算法是
 *		桶号 = hashvalue MOD nbuckets
 *		批次号 = ROR(hashvalue, log2_nbuckets) MOD nbatch
 * 其中 nbuckets 和 nbatch 都应该是 2 的幂，这样我们就可以 
 * 通过位移和掩码来进行计算。（这假设所有哈希 
 * 函数在随机化所有输出位方面的表现良好，否则我们可能会有非常偏斜的桶或批次占用。）
 *
 * 在 nbatch == 1 时，nbuckets 和 log2_nbuckets 可能会发生变化，因为动态 
 * 增加了桶的数量。一旦我们开始分批，值是固定的，在连接过程中不会变化 
 * （这使得我们可以按照这里所述的方式计算批次号）。
 *
 * nbatch 总是 2 的幂；我们只通过将其加倍来增加它。 这
 * 实际上是在批次号的顶部添加了一个比特。在非常大的 
 * 连接中，我们可能会用尽可以添加的位，因此我们通过旋转哈希 
 * 值来解决这个问题。这会导致批次号从桶号中窃取位数，当虚拟桶的数量超过 2^32 时出现。 
 * 拥有更长的桶链比失去分批能力要好得多。
 */
void ExecHashGetBucketAndBatch(HashJoinTable hashtable,
						  uint32 fc_hashvalue,
						  int *fc_bucketno,
						  int *fc_batchno)
{
	uint32		fc_nbuckets = (uint32) hashtable->nbuckets;
	uint32		fc_nbatch = (uint32) hashtable->nbatch;

	if (fc_nbatch > 1)
	{
		*fc_bucketno = fc_hashvalue & (fc_nbuckets - 1);
		*fc_batchno = pg_rotate_right32(fc_hashvalue,
									 hashtable->log2_nbuckets) & (fc_nbatch - 1);
	}
	else
	{
		*fc_bucketno = fc_hashvalue & (fc_nbuckets - 1);
		*fc_batchno = 0;
	}
}

/* 
 * ExecScanHashBucket
 *		扫描哈希桶以匹配当前外部元组
 *
 * 当前外部元组必须存储在 econtext->ecxt_outertuple 中。
 *
 * 成功时，内部元组存储在 hjstate->hj_CurTuple 和 
 * econtext->ecxt_innertuple 中，使用 hjstate->hj_HashTupleSlot 作为后者的槽。
 */
bool ExecScanHashBucket(HashJoinState *fc_hjstate,
				   ExprContext *fc_econtext)
{
	ExprState  *fc_hjclauses = fc_hjstate->hashclauses;
	HashJoinTable hashtable = fc_hjstate->hj_HashTable;
	HashJoinTuple fc_hashTuple = fc_hjstate->hj_CurTuple;
	uint32		fc_hashvalue = fc_hjstate->hj_CurHashValue;

	/* 
	 * hj_CurTuple 是当前 
	 * 桶中最后返回的元组的地址，如果是时候开始扫描一个新桶，则为 NULL。
	 *
	 * 如果元组哈希到偏斜桶，则扫描偏斜桶 
	 * 否则扫描标准哈希表桶。
	 */
	if (fc_hashTuple != NULL)
		fc_hashTuple = fc_hashTuple->next.unshared;
	else if (fc_hjstate->hj_CurSkewBucketNo != INVALID_SKEW_BUCKET_NO)
		fc_hashTuple = hashtable->skewBucket[fc_hjstate->hj_CurSkewBucketNo]->tuples;
	else
		fc_hashTuple = hashtable->buckets.unshared[fc_hjstate->hj_CurBucketNo];

	while (fc_hashTuple != NULL)
	{
		if (fc_hashTuple->hashvalue == fc_hashvalue)
		{
			TupleTableSlot *fc_inntuple;

			/* 将哈希表的元组插入执行槽，以便 ExecQual 可以看到它 */
			fc_inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(fc_hashTuple),
											 fc_hjstate->hj_HashTupleSlot,
											 false);	/* 不要释放 */
			fc_econtext->ecxt_innertuple = fc_inntuple;

			if (ExecQualAndReset(fc_hjclauses, fc_econtext))
			{
				fc_hjstate->hj_CurTuple = fc_hashTuple;
				return true;
			}
		}

		fc_hashTuple = fc_hashTuple->next.unshared;
	}

	/* 
	 * 没有匹配项
	 */
	return false;
}

/* 
 * ExecParallelScanHashBucket
 *		扫描哈希桶以匹配当前外部元组
 *
 * 当前外部元组必须存储在 econtext->ecxt_outertuple 中。
 *
 * 成功时，内部元组存储在 hjstate->hj_CurTuple 和 
 * econtext->ecxt_innertuple 中，使用 hjstate->hj_HashTupleSlot 作为后者的槽。
 */
bool ExecParallelScanHashBucket(HashJoinState *fc_hjstate,
						   ExprContext *fc_econtext)
{
	ExprState  *fc_hjclauses = fc_hjstate->hashclauses;
	HashJoinTable hashtable = fc_hjstate->hj_HashTable;
	HashJoinTuple fc_hashTuple = fc_hjstate->hj_CurTuple;
	uint32		fc_hashvalue = fc_hjstate->hj_CurHashValue;

	/* 
	 * hj_CurTuple 是当前 
	 * 桶中最后返回的元组的地址，如果是时候开始扫描一个新桶，则为 NULL。
	 */
	if (fc_hashTuple != NULL)
		fc_hashTuple = fc_ExecParallelHashNextTuple(hashtable, fc_hashTuple);
	else
		fc_hashTuple = fc_ExecParallelHashFirstTuple(hashtable,
											   fc_hjstate->hj_CurBucketNo);

	while (fc_hashTuple != NULL)
	{
		if (fc_hashTuple->hashvalue == fc_hashvalue)
		{
			TupleTableSlot *fc_inntuple;

			/* 将哈希表的元组插入执行槽，以便 ExecQual 可以看到它 */
			fc_inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(fc_hashTuple),
											 fc_hjstate->hj_HashTupleSlot,
											 false);	/* 不要释放 */
			fc_econtext->ecxt_innertuple = fc_inntuple;

			if (ExecQualAndReset(fc_hjclauses, fc_econtext))
			{
				fc_hjstate->hj_CurTuple = fc_hashTuple;
				return true;
			}
		}

		fc_hashTuple = fc_ExecParallelHashNextTuple(hashtable, fc_hashTuple);
	}

	/* 
	 * 没有匹配项
	 */
	return false;
}

/* 
 * ExecPrepHashTableForUnmatched
 *		为一系列 ExecScanHashTableForUnmatched 调用做好准备
 */
void ExecPrepHashTableForUnmatched(HashJoinState *fc_hjstate)
{
	/*----------
	 * 在此扫描中，我们使用 HashJoinState 字段如下：
	 *
	 * hj_CurBucketNo: 下一个常规桶以进行扫描
	 * hj_CurSkewBucketNo: 下一个偏斜桶（偏斜桶数量的索引）
	 * hj_CurTuple: 上一个返回的元组，或 NULL 以开始下一个桶
	 *----------
	 */
	fc_hjstate->hj_CurBucketNo = 0;
	fc_hjstate->hj_CurSkewBucketNo = 0;
	fc_hjstate->hj_CurTuple = NULL;
}

/* 
 * ExecScanHashTableForUnmatched
 *		扫描哈希表以查找未匹配的内部元组
 *
 * 成功时，内部元组存储在 hjstate->hj_CurTuple 和 
 * econtext->ecxt_innertuple 中，使用 hjstate->hj_HashTupleSlot 作为后者的槽。
 */
bool ExecScanHashTableForUnmatched(HashJoinState *fc_hjstate, ExprContext *fc_econtext)
{
	HashJoinTable hashtable = fc_hjstate->hj_HashTable;
	HashJoinTuple fc_hashTuple = fc_hjstate->hj_CurTuple;

	for (;;)
	{
		/* 
		 * hj_CurTuple 是最后一个返回的元组的地址 
		 * 来自当前桶，如果是时候开始扫描一个新 
		 * 桶，则为 NULL。
		 */
		if (fc_hashTuple != NULL)
			fc_hashTuple = fc_hashTuple->next.unshared;
		else if (fc_hjstate->hj_CurBucketNo < hashtable->nbuckets)
		{
			fc_hashTuple = hashtable->buckets.unshared[fc_hjstate->hj_CurBucketNo];
			fc_hjstate->hj_CurBucketNo++;
		}
		else if (fc_hjstate->hj_CurSkewBucketNo < hashtable->nSkewBuckets)
		{
			int			fc_j = hashtable->skewBucketNums[fc_hjstate->hj_CurSkewBucketNo];

			fc_hashTuple = hashtable->skewBucket[fc_j]->tuples;
			fc_hjstate->hj_CurSkewBucketNo++;
		}
		else
			break;				/* 已完成所有桶 */

		while (fc_hashTuple != NULL)
		{
			if (!HeapTupleHeaderHasMatch(HJTUPLE_MINTUPLE(fc_hashTuple)))
			{
				TupleTableSlot *fc_inntuple;

				/* 将哈希表的元组插入执行槽 */
				fc_inntuple = ExecStoreMinimalTuple(HJTUPLE_MINTUPLE(fc_hashTuple),
												 fc_hjstate->hj_HashTupleSlot,
												 false);	/* 不要释放 */
				fc_econtext->ecxt_innertuple = fc_inntuple;

				/* 
				 * 每次重置临时内存；尽管此函数不
				 * 执行任何条件评估，但调用者会这样做，因此我们保持它
				 * 与 ExecScanHashBucket 并行。
				 */
				ResetExprContext(fc_econtext);

				fc_hjstate->hj_CurTuple = fc_hashTuple;
				return true;
			}

			fc_hashTuple = fc_hashTuple->next.unshared;
		}

		/* 允许此循环可取消 */
		CHECK_FOR_INTERRUPTS();
	}

	/* 
	 * 没有更多未匹配的元组
	 */
	return false;
}

/* 
 * ExecHashTableReset
 *
 *		为新批次重置哈希表头
 */
void ExecHashTableReset(HashJoinTable hashtable)
{
	MemoryContext fc_oldcxt;
	int			fc_nbuckets = hashtable->nbuckets;

	/* 
	 * 释放先前传递中获取的所有哈希桶和元组，并
	 * 为新传递重新初始化上下文。
	 */
	MemoryContextReset(hashtable->batchCxt);
	fc_oldcxt = MemoryContextSwitchTo(hashtable->batchCxt);

	/* 重新分配和重新初始化哈希桶头。 */
	hashtable->buckets.unshared = (HashJoinTuple *)
		palloc0(fc_nbuckets * sizeof(HashJoinTuple));

	hashtable->spaceUsed = 0;

	MemoryContextSwitchTo(fc_oldcxt);

	/* 忘记块（以上的内存已被上下文重置释放）。 */
	hashtable->chunks = NULL;
}

/* 
 * ExecHashTableResetMatchFlags
 *		清除表中的所有 HeapTupleHeaderHasMatch 标志
 */
void ExecHashTableResetMatchFlags(HashJoinTable hashtable)
{
	HashJoinTuple fc_tuple;
	int			fc_i;

	/* 重置主表中的所有标志... */
	for (fc_i = 0; fc_i < hashtable->nbuckets; fc_i++)
	{
		for (fc_tuple = hashtable->buckets.unshared[fc_i]; fc_tuple != NULL;
			 fc_tuple = fc_tuple->next.unshared)
			HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(fc_tuple));
	}

	/* ...如果有的话，对偏斜桶也如此 */
	for (fc_i = 0; fc_i < hashtable->nSkewBuckets; fc_i++)
	{
		int			fc_j = hashtable->skewBucketNums[fc_i];
		HashSkewBucket *fc_skewBucket = hashtable->skewBucket[fc_j];

		for (fc_tuple = fc_skewBucket->tuples; fc_tuple != NULL; fc_tuple = fc_tuple->next.unshared)
			HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(fc_tuple));
	}
}


void ExecReScanHash(HashState *fc_node)
{
	/*
	 * 如果子节点的chgParam不为空，则计划将由第一个ExecProcNode重新扫描。
	 */
	if (fc_node->ps.lefttree->chgParam == NULL)
		ExecReScan(fc_node->ps.lefttree);
}


/* 
 * ExecHashBuildSkewHash
 *
 *		如果我们能够识别外部关系的连接键的最常见值（MCV），则为偏斜优化做好准备。
 *		我们为每个 MCV 的哈希值制作一个偏斜哈希桶，数量取决于可用内存的槽位。
 */
static void fc_ExecHashBuildSkewHash(HashJoinTable hashtable, Hash *fc_node, int fc_mcvsToUse)
{
	HeapTupleData *fc_statsTuple;
	AttStatsSlot fc_sslot;

	
/* 如果规划器没有识别出外部关系的连接键，则不进行任何操作 */
	if (!OidIsValid(fc_node->skewTable))
		return;
	/* 如果我们没有空间容纳至少一个偏斜桶，也不进行任何操作 */
	if (fc_mcvsToUse <= 0)
		return;

	/*
	 * 尝试找到外部关系连接键的 MCV 统计信息。
	 */
	fc_statsTuple = SearchSysCache3(STATRELATTINH,
								 ObjectIdGetDatum(fc_node->skewTable),
								 Int16GetDatum(fc_node->skewColumn),
								 BoolGetDatum(fc_node->skewInherit));
	if (!HeapTupleIsValid(fc_statsTuple))
		return;

	if (get_attstatsslot(&fc_sslot, fc_statsTuple,
						 STATISTIC_KIND_MCV, InvalidOid,
						 ATTSTATSSLOT_VALUES | ATTSTATSSLOT_NUMBERS))
	{
		double		fc_frac;
		int			fc_nbuckets;
		FmgrInfo   *fc_hashfunctions;
		int			fc_i;

		if (fc_mcvsToUse > fc_sslot.nvalues)
			fc_mcvsToUse = fc_sslot.nvalues;

		/*
		 * 计算参与偏斜优化的外部关系的预期比例。如果这比例不足
		 * SKEW_MIN_OUTER_FRACTION，则不使用偏斜优化。
		 */
		fc_frac = 0;
		for (fc_i = 0; fc_i < fc_mcvsToUse; fc_i++)
			fc_frac += fc_sslot.numbers[fc_i];
		if (fc_frac < SKEW_MIN_OUTER_FRACTION)
		{
			free_attstatsslot(&fc_sslot);
			ReleaseSysCache(fc_statsTuple);
			return;
		}

		/*
		 * 好的，设置偏斜哈希表。
		 *
		 * skewBucket[] 是一个开放寻址的哈希表，其大小为 2 的幂，
		 * 大于 MCV 值的数量。 （这确保至少会有一个空条目，因此搜索
		 * 将始终终止。）
		 *
		 * 注意：如果 mcvsToUse 超过 INT_MAX/8 或 MaxAllocSize/sizeof(void *)/8，
		 * 这段代码可能会失败，但由于我们将 pg_statistic 条目限制在远低于此，
		 * 所以目前不可能发生这种情况。
		 */
		fc_nbuckets = pg_nextpower2_32(fc_mcvsToUse + 1);
		/* 再使用两个比特以帮助避免冲突 */
		fc_nbuckets <<= 2;

		hashtable->skewEnabled = true;
		hashtable->skewBucketLen = fc_nbuckets;

		/*
		 * 我们在哈希表的批处理上下文中分配桶的内存。它仅在第一次批处理中需要，
		 * 这确保它在第一次批处理完成后会被自动移除。
		 */
		hashtable->skewBucket = (HashSkewBucket **)
			MemoryContextAllocZero(hashtable->batchCxt,
								   fc_nbuckets * sizeof(HashSkewBucket *));
		hashtable->skewBucketNums = (int *)
			MemoryContextAllocZero(hashtable->batchCxt,
								   fc_mcvsToUse * sizeof(int));

		hashtable->spaceUsed += fc_nbuckets * sizeof(HashSkewBucket *)
			+ fc_mcvsToUse * sizeof(int);
		hashtable->spaceUsedSkew += fc_nbuckets * sizeof(HashSkewBucket *)
			+ fc_mcvsToUse * sizeof(int);
		if (hashtable->spaceUsed > hashtable->spacePeak)
			hashtable->spacePeak = hashtable->spaceUsed;

		/*
		 * 为每个 MCV 哈希值创建一个偏斜桶。
		 *
		 * 注意：非常重要的是，我们按 decreasing MCV 频率的顺序创建桶。
		 * 如果我们不得不移除一些桶，它们必须按创建的反向顺序移除（请参见
		 * ExecHashRemoveNextSkewBucket 中的说明），我们希望最少见的 MCV
		 * 先被移除。
		 */
		fc_hashfunctions = hashtable->outer_hashfunctions;

		for (fc_i = 0; fc_i < fc_mcvsToUse; fc_i++)
		{
			uint32		fc_hashvalue;
			int			fc_bucket;

			fc_hashvalue = DatumGetUInt32(FunctionCall1Coll(&fc_hashfunctions[0],
														 hashtable->collations[0],
														 fc_sslot.values[fc_i]));

			/*
			 * 当我们没有遇到哈希表中的空桶，并且还没有命中
			 * 所需的桶时，我们与一些以前的哈希值发生了冲突，
			 * 所以尝试下一个桶的位置。注意：这段代码必须
			 * 与 ExecHashGetSkewBucket 匹配。
			 */
			fc_bucket = fc_hashvalue & (fc_nbuckets - 1);
			while (hashtable->skewBucket[fc_bucket] != NULL &&
				   hashtable->skewBucket[fc_bucket]->hashvalue != fc_hashvalue)
				fc_bucket = (fc_bucket + 1) & (fc_nbuckets - 1);

			/*
			 * 如果我们发现一个具有相同 hashvalue 的现有桶，保持
			 * 它不变。两个 MCV 共享一个 hashvalue 是可以的。
			 */
			if (hashtable->skewBucket[fc_bucket] != NULL)
				continue;

			/* 好的，为这个 hashvalue 创建一个新的偏斜桶。 */
			hashtable->skewBucket[fc_bucket] = (HashSkewBucket *)
				MemoryContextAlloc(hashtable->batchCxt,
								   sizeof(HashSkewBucket));
			hashtable->skewBucket[fc_bucket]->hashvalue = fc_hashvalue;
			hashtable->skewBucket[fc_bucket]->tuples = NULL;
			hashtable->skewBucketNums[hashtable->nSkewBuckets] = fc_bucket;
			hashtable->nSkewBuckets++;
			hashtable->spaceUsed += SKEW_BUCKET_OVERHEAD;
			hashtable->spaceUsedSkew += SKEW_BUCKET_OVERHEAD;
			if (hashtable->spaceUsed > hashtable->spacePeak)
				hashtable->spacePeak = hashtable->spaceUsed;
		}

		free_attstatsslot(&fc_sslot);
	}

	ReleaseSysCache(fc_statsTuple);
}

/*
 * ExecHashGetSkewBucket
 *
 *		返回此 hashvalue 的偏斜桶的索引，
 *		如果该 hashvalue 不与任何活动的偏斜桶相关联，则返回 INVALID_SKEW_BUCKET_NO。
 */
int ExecHashGetSkewBucket(HashJoinTable hashtable, uint32 fc_hashvalue)
{
	int			fc_bucket;

	/*
	 * 如果未进行偏斜优化，则始终返回 INVALID_SKEW_BUCKET_NO
	 * （特别是，这在初始批处理完成后发生）。
	 */
	if (!hashtable->skewEnabled)
		return INVALID_SKEW_BUCKET_NO;

	/*
	 * 由于 skewBucketLen 是 2 的幂，我们可以通过 AND 运算进行取模。
	 */
	fc_bucket = fc_hashvalue & (hashtable->skewBucketLen - 1);

	/*
	 * 当我们没有遇到哈希表中的空桶，并且还没有命中
	 * 所需的桶时，我们与另一个哈希值发生了冲突，所以尝试下一个
	 * 桶的位置。
	 */
	while (hashtable->skewBucket[fc_bucket] != NULL &&
		   hashtable->skewBucket[fc_bucket]->hashvalue != fc_hashvalue)
		fc_bucket = (fc_bucket + 1) & (hashtable->skewBucketLen - 1);

	/*
	 * 找到所需的桶了吗？
	 */
	if (hashtable->skewBucket[fc_bucket] != NULL)
		return fc_bucket;

	/*
	 * 该 hash value 不能有任何哈希表条目。
	 */
	return INVALID_SKEW_BUCKET_NO;
}

/*
 * ExecHashSkewTableInsert
 *
 *		将元组插入偏斜哈希表。
 *
 * 这通常应与 ExecHashTableInsert 中的当前批处理情况相匹配。
 */
static void fc_ExecHashSkewTableInsert(HashJoinTable hashtable,
						TupleTableSlot *fc_slot,
						uint32 fc_hashvalue,
						int fc_bucketNumber)
{
	bool		fc_shouldFree;
	MinimalTuple fc_tuple = ExecFetchSlotMinimalTuple(fc_slot, &fc_shouldFree);
	HashJoinTuple fc_hashTuple;
	int			fc_hashTupleSize;

	/* 创建 HashJoinTuple */
	fc_hashTupleSize = HJTUPLE_OVERHEAD + fc_tuple->t_len;
	fc_hashTuple = (HashJoinTuple) MemoryContextAlloc(hashtable->batchCxt,
												   fc_hashTupleSize);
	fc_hashTuple->hashvalue = fc_hashvalue;
	memcpy(HJTUPLE_MINTUPLE(fc_hashTuple), fc_tuple, fc_tuple->t_len);
	HeapTupleHeaderClearMatch(HJTUPLE_MINTUPLE(fc_hashTuple));

	/* 将其推到偏斜桶列表的前面 */
	fc_hashTuple->next.unshared = hashtable->skewBucket[fc_bucketNumber]->tuples;
	hashtable->skewBucket[fc_bucketNumber]->tuples = fc_hashTuple;
	Assert(fc_hashTuple != fc_hashTuple->next.unshared);

	/* 计算使用的空间，如果使用过多则退回 */
	hashtable->spaceUsed += fc_hashTupleSize;
	hashtable->spaceUsedSkew += fc_hashTupleSize;
	if (hashtable->spaceUsed > hashtable->spacePeak)
		hashtable->spacePeak = hashtable->spaceUsed;
	while (hashtable->spaceUsedSkew > hashtable->spaceAllowedSkew)
		fc_ExecHashRemoveNextSkewBucket(hashtable);

	/* 检查我们是否没有超出总空间允许的限制 */
	if (hashtable->spaceUsed > hashtable->spaceAllowed)
		fc_ExecHashIncreaseNumBatches(hashtable);

	if (fc_shouldFree)
		heap_free_minimal_tuple(fc_tuple);
}

/*
 *		ExecHashRemoveNextSkewBucket
 *
 *		通过将其元组推入主哈希表来移除最少价值的偏斜桶。
 */
static void fc_ExecHashRemoveNextSkewBucket(HashJoinTable hashtable)
{
	int			fc_bucketToRemove;
	HashSkewBucket *fc_bucket;
	uint32		fc_hashvalue;
	int			fc_bucketno;
	int			fc_batchno;
	HashJoinTuple fc_hashTuple;

	/* 定位要移除的桶 */
	fc_bucketToRemove = hashtable->skewBucketNums[hashtable->nSkewBuckets - 1];
	fc_bucket = hashtable->skewBucket[fc_bucketToRemove];

	/*
	 * 计算元组在主哈希表中所属的桶和批次。它们都具有相同的哈希值，
	 * 所以对它们而言都是相同的。还要注意，在处理元组时 nbatch
	 * 不可能增加。
	 */
	fc_hashvalue = fc_bucket->hashvalue;
	ExecHashGetBucketAndBatch(hashtable, fc_hashvalue, &fc_bucketno, &fc_batchno);

	/* 处理桶中的所有元组 */
	fc_hashTuple = fc_bucket->tuples;
	while (fc_hashTuple != NULL)
	{
		HashJoinTuple fc_nextHashTuple = fc_hashTuple->next.unshared;
		MinimalTuple fc_tuple;
		Size		fc_tupleSize;

		/*
		 * 这段代码必须与 ExecHashTableInsert 一致。我们不直接使用
		 * ExecHashTableInsert，因为 ExecHashTableInsert 期望一个
		 * TupleTableSlot，而我们已经拥有 HashJoinTuples。
		 */
		fc_tuple = HJTUPLE_MINTUPLE(fc_hashTuple);
		fc_tupleSize = HJTUPLE_OVERHEAD + fc_tuple->t_len;

		/* 决定是将元组放入哈希表中还是放入临时文件中 */
		if (fc_batchno == hashtable->curbatch)
		{
			/* 将元组移至主哈希表 */
			HashJoinTuple fc_copyTuple;

			/*
			 * 我们必须将元组复制到密集存储中，否则它不会被
			 * 例如，ExecHashIncreaseNumBatches 找到。
			 */
			fc_copyTuple = (HashJoinTuple) fc_dense_alloc(hashtable, fc_tupleSize);
			memcpy(fc_copyTuple, fc_hashTuple, fc_tupleSize);
			pfree(fc_hashTuple);

			fc_copyTuple->next.unshared = hashtable->buckets.unshared[fc_bucketno];
			hashtable->buckets.unshared[fc_bucketno] = fc_copyTuple;

			/* 我们已经减少了偏斜空间，但总体空间没有变化 */
			hashtable->spaceUsedSkew -= fc_tupleSize;
		}
		else
		{
			/* 将元组放入临时文件以便于后续批处理 */
			Assert(fc_batchno > hashtable->curbatch);
			ExecHashJoinSaveTuple(fc_tuple, fc_hashvalue,
								  &hashtable->innerBatchFile[fc_batchno]);
			pfree(fc_hashTuple);
			hashtable->spaceUsed -= fc_tupleSize;
			hashtable->spaceUsedSkew -= fc_tupleSize;
		}

		fc_hashTuple = fc_nextHashTuple;

		/* 允许此循环可取消 */
		CHECK_FOR_INTERRUPTS();
	}

	/*
	 * 释放桶的结构本身并将哈希表条目重置为 NULL。
	 *
	 * 注意：这并没有表面上看起来的那么简单，因为
	 * 哈希表中可能会发生冲突。假设哈希值 A 和 B 在特定的哈希表条目上发生冲突，
	 * 并且 A 首先被输入，因此 B 被移到不同的表条目。如果我们首先
	 * 移除 A，则 ExecHashGetSkewBucket 会错误地开始报告 B 不在
	 * 哈希表中，因为它会在找到 B 之前遇到 NULL。然而，我们总是按创建的反向顺序移除条目，
	 * 因此这种失败不能发生。
	 */
	hashtable->skewBucket[fc_bucketToRemove] = NULL;
	hashtable->nSkewBuckets--;
	pfree(fc_bucket);
	hashtable->spaceUsed -= SKEW_BUCKET_OVERHEAD;
	hashtable->spaceUsedSkew -= SKEW_BUCKET_OVERHEAD;

	/*
	 * 如果我们已经移除了所有的偏斜桶，则放弃偏斜优化。
	 * 释放数组，因为它们不再有用。
	 */
	if (hashtable->nSkewBuckets == 0)
	{
		hashtable->skewEnabled = false;
		pfree(hashtable->skewBucket);
		pfree(hashtable->skewBucketNums);
		hashtable->skewBucket = NULL;
		hashtable->skewBucketNums = NULL;
		hashtable->spaceUsed -= hashtable->spaceUsedSkew;
		hashtable->spaceUsedSkew = 0;
	}
}

/*
 * 在 DSM 段中为仪器数据保留空间。
 */
void ExecHashEstimate(HashState *fc_node, ParallelContext *fc_pcxt)
{
	size_t		fc_size;

	/* 如果没有工具监控或没有工作程序，则不需要此 */
	if (!fc_node->ps.instrument || fc_pcxt->nworkers == 0)
		return;

	fc_size = mul_size(fc_pcxt->nworkers, sizeof(HashInstrumentation));
	fc_size = add_size(fc_size, offsetof(SharedHashInfo, hinstrument));
	shm_toc_estimate_chunk(&fc_pcxt->estimator, fc_size);
	shm_toc_estimate_keys(&fc_pcxt->estimator, 1);
}

/*
 * 为所有工作线程在DSM中设置一个空间，以记录关于它们的哈希表的仪器数据。
 */
void ExecHashInitializeDSM(HashState *fc_node, ParallelContext *fc_pcxt)
{
	size_t		fc_size;

	/* 如果没有工具监控或没有工作程序，则不需要此 */
	if (!fc_node->ps.instrument || fc_pcxt->nworkers == 0)
		return;

	fc_size = offsetof(SharedHashInfo, hinstrument) +
		fc_pcxt->nworkers * sizeof(HashInstrumentation);
	fc_node->shared_info = (SharedHashInfo *) shm_toc_allocate(fc_pcxt->toc, fc_size);

	/* 每个工作线程区域必须最初为零。 */
	memset(fc_node->shared_info, 0, fc_size);

	fc_node->shared_info->num_workers = fc_pcxt->nworkers;
	shm_toc_insert(fc_pcxt->toc, fc_node->ps.plan->plan_node_id,
				   fc_node->shared_info);
}

/*
 * 定位我们将在关闭时写入的哈希表仪器数据的DSM空间。
 */
void ExecHashInitializeWorker(HashState *fc_node, ParallelWorkerContext *fc_pwcxt)
{
	SharedHashInfo *fc_shared_info;

	/* 如果不进行仪器测量，则不需要此项。 */
	if (!fc_node->ps.instrument)
		return;

	/*
	 * 在共享区域中找到我们的条目，并设置一个指针，以便我们
	 * 在关闭或重建哈希表时在此收集统计信息。
	 */
	fc_shared_info = (SharedHashInfo *)
		shm_toc_lookup(fc_pwcxt->toc, fc_node->ps.plan->plan_node_id, false);
	fc_node->hinstrument = &fc_shared_info->hinstrument[ParallelWorkerNumber];
}

/*
 * 如果需要，收集EXPLAIN统计信息，如果调用了ExecHashInitializeWorker，
 * 将其保存在DSM内存中；否则保存在本地存储中。在并行情况下，这必须在
 * ExecShutdownHash()中完成，而不是ExecEndHash()，因为后者在我们断开与DSM
 * 段的连接后运行。
 */
void ExecShutdownHash(HashState *fc_node)
{
	/* 如果正在进行EXPLAIN且尚未分配保存空间，则分配。 */
	if (fc_node->ps.instrument && !fc_node->hinstrument)
		fc_node->hinstrument = (HashInstrumentation *)
			palloc0(sizeof(HashInstrumentation));
	/* 现在收集当前（最终）哈希表的数据。 */
	if (fc_node->hinstrument && fc_node->hashtable)
		ExecHashAccumInstrumentation(fc_node->hinstrument, fc_node->hashtable);
}

/*
 * 在DSM段被分离之前，从工作线程检索仪器数据，以便EXPLAIN可以访问它。
 */
void ExecHashRetrieveInstrumentation(HashState *fc_node)
{
	SharedHashInfo *fc_shared_info = fc_node->shared_info;
	size_t		fc_size;

	if (fc_shared_info == NULL)
		return;

	/* 用后端本地内存中的副本替换node->shared_info。 */
	fc_size = offsetof(SharedHashInfo, hinstrument) +
		fc_shared_info->num_workers * sizeof(HashInstrumentation);
	fc_node->shared_info = palloc(fc_size);
	memcpy(fc_node->shared_info, fc_shared_info, fc_size);
}

/*
 * 从'hashtable'中收集仪器数据到一个最初为零的HashInstrumentation结构中。
 *
 * 这用于合并单个计划节点中连续哈希表实例的信息。我们取每个有趣
 * 数字的最大值。最大nbuckets和最大nbatch值可能出现在不同实例中，
 * 因此报告不相关数字存在一些混淆的风险；但如果我们不报告最大值，
 * 可能会更大风险地误诊性能问题。类似地，我们希望报告最大spacePeak
 * ，而不管它是否发生在与最大nbuckets或nbatch相同的实例中。所有实例
 * 应该具有相同的nbuckets_original和nbatch_original；但在这里依赖于
 * 这一点没有太大价值，因此以相同方式处理它们。
 */
void ExecHashAccumInstrumentation(HashInstrumentation *fc_instrument,
							 HashJoinTable hashtable)
{
	fc_instrument->nbuckets = Max(fc_instrument->nbuckets,
							   hashtable->nbuckets);
	fc_instrument->nbuckets_original = Max(fc_instrument->nbuckets_original,
										hashtable->nbuckets_original);
	fc_instrument->nbatch = Max(fc_instrument->nbatch,
							 hashtable->nbatch);
	fc_instrument->nbatch_original = Max(fc_instrument->nbatch_original,
									  hashtable->nbatch_original);
	fc_instrument->space_peak = Max(fc_instrument->space_peak,
								 hashtable->spacePeak);
}

/*
 * 从当前活动的HashMemoryChunk中分配'size'字节。
 */
static void * fc_dense_alloc(HashJoinTable hashtable, Size fc_size)
{
	HashMemoryChunk fc_newChunk;
	char	   *fc_ptr;

	/* 以防大小尚未正确对齐。 */
	fc_size = MAXALIGN(fc_size);

	/*
	 * 如果元组大小大于阈值，则分配一个单独的块。
	 */
	if (fc_size > HASH_CHUNK_THRESHOLD)
	{
		/* 分配新块，并将其放在列表的开头。 */
		fc_newChunk = (HashMemoryChunk) MemoryContextAlloc(hashtable->batchCxt,
														HASH_CHUNK_HEADER_SIZE + fc_size);
		fc_newChunk->maxlen = fc_size;
		fc_newChunk->used = fc_size;
		fc_newChunk->ntuples = 1;

		/*
		 * 在第一个现有块后将此块添加到列表中，以便我们不会丢失
		 * “当前”块中剩余的空间。
		 */
		if (hashtable->chunks != NULL)
		{
			fc_newChunk->next = hashtable->chunks->next;
			hashtable->chunks->next.unshared = fc_newChunk;
		}
		else
		{
			fc_newChunk->next.unshared = hashtable->chunks;
			hashtable->chunks = fc_newChunk;
		}

		return HASH_CHUNK_DATA(fc_newChunk);
	}

	/*
	 * 查看在当前块中是否有足够的空间（如果有的话）。如果没有，
	 * 则分配一个新的块。
	 */
	if ((hashtable->chunks == NULL) ||
		(hashtable->chunks->maxlen - hashtable->chunks->used) < fc_size)
	{
		/* 分配新块，并将其放在列表的开头。 */
		fc_newChunk = (HashMemoryChunk) MemoryContextAlloc(hashtable->batchCxt,
														HASH_CHUNK_HEADER_SIZE + HASH_CHUNK_SIZE);

		fc_newChunk->maxlen = HASH_CHUNK_SIZE;
		fc_newChunk->used = fc_size;
		fc_newChunk->ntuples = 1;

		fc_newChunk->next.unshared = hashtable->chunks;
		hashtable->chunks = fc_newChunk;

		return HASH_CHUNK_DATA(fc_newChunk);
	}

	/* 当前块中有足够的空间，让我们添加元组。 */
	fc_ptr = HASH_CHUNK_DATA(hashtable->chunks) + hashtable->chunks->used;
	hashtable->chunks->used += fc_size;
	hashtable->chunks->ntuples += 1;

	/* 返回指向元组内存开始的指针。 */
	return fc_ptr;
}

/*
 * 在共享稠密存储中为元组分配空间。这相当于dense_alloc，但用于
 * 使用共享内存的Parallel Hash。
 *
 * 在将元组加载到共享内存时，我们可能会耗尽内存并决定重新分区，或
 * 判定负载因子过高而决定扩展桶数组，或发现另一个参与者命令我们
 * 帮助执行该操作。如果桶数或批次数发生变化，则返回NULL，表示调用
 * 者必须重试（考虑到元组不再属于同一批次的可能性）。
 */
static HashJoinTuple
fc_ExecParallelHashTupleAlloc(HashJoinTable hashtable, size_t fc_size,
						   dsa_pointer *fc_shared)
{
	ParallelHashJoinState *fc_pstate = hashtable->parallel_state;
	dsa_pointer fc_chunk_shared;
	HashMemoryChunk fc_chunk;
	Size		fc_chunk_size;
	HashJoinTuple fc_result;
	int			fc_curbatch = hashtable->curbatch;

	fc_size = MAXALIGN(fc_size);

	/*
	 * 快速路径：如果这个后端的当前块中有足够的空间，
	 * 那么我们可以在没有任何锁定的情况下进行分配。
	 */
	fc_chunk = hashtable->current_chunk;
	if (fc_chunk != NULL &&
		fc_size <= HASH_CHUNK_THRESHOLD &&
		fc_chunk->maxlen - fc_chunk->used >= fc_size)
	{

		fc_chunk_shared = hashtable->current_chunk_shared;
		Assert(fc_chunk == dsa_get_address(hashtable->area, fc_chunk_shared));
		*fc_shared = fc_chunk_shared + HASH_CHUNK_HEADER_SIZE + fc_chunk->used;
		fc_result = (HashJoinTuple) (HASH_CHUNK_DATA(fc_chunk) + fc_chunk->used);
		fc_chunk->used += fc_size;

		Assert(fc_chunk->used <= fc_chunk->maxlen);
		Assert(fc_result == dsa_get_address(hashtable->area, *fc_shared));

		return fc_result;
	}

	/* 慢慢路径：尝试分配一个新的块。 */
	LWLockAcquire(&fc_pstate->lock, LW_EXCLUSIVE);

	/*
	 * 检查是否需要帮助增加桶数或批次数。
	 */
	if (fc_pstate->growth == PHJ_GROWTH_NEED_MORE_BATCHES ||
		fc_pstate->growth == PHJ_GROWTH_NEED_MORE_BUCKETS)
	{
		ParallelHashGrowth fc_growth = fc_pstate->growth;

		hashtable->current_chunk = NULL;
		LWLockRelease(&fc_pstate->lock);

		/* 另一个参与者命令我们帮助扩展。 */
		if (fc_growth == PHJ_GROWTH_NEED_MORE_BATCHES)
			fc_ExecParallelHashIncreaseNumBatches(hashtable);
		else if (fc_growth == PHJ_GROWTH_NEED_MORE_BUCKETS)
			fc_ExecParallelHashIncreaseNumBuckets(hashtable);

		/* 调用者必须重试。 */
		return NULL;
	}

	/* 超大元组获得自己的块。 */
	if (fc_size > HASH_CHUNK_THRESHOLD)
		fc_chunk_size = fc_size + HASH_CHUNK_HEADER_SIZE;
	else
		fc_chunk_size = HASH_CHUNK_SIZE;

	/* 检查是否该扩展批次或桶。 */
	if (fc_pstate->growth != PHJ_GROWTH_DISABLED)
	{
		Assert(fc_curbatch == 0);
		Assert(BarrierPhase(&fc_pstate->build_barrier) == PHJ_BUILD_HASHING_INNER);

		/*
		 * 检查我们的空间限制是否会被超过。为了避免在非常大
		 * 的元组或非常低的hash_mem设置下阻塞，我们将始终允许
		 * 每个后端至少分配一个块。
		 */
		if (hashtable->batches[0].at_least_one_chunk &&
			hashtable->batches[0].shared->size +
			fc_chunk_size > fc_pstate->space_allowed)
		{
			fc_pstate->growth = PHJ_GROWTH_NEED_MORE_BATCHES;
			hashtable->batches[0].shared->space_exhausted = true;
			LWLockRelease(&fc_pstate->lock);

			return NULL;
		}

		/* 检查我们的负载因子限制是否会被超过。 */
		if (hashtable->nbatch == 1)
		{
			hashtable->batches[0].shared->ntuples += hashtable->batches[0].ntuples;
			hashtable->batches[0].ntuples = 0;
			/* 防止整数溢出和分配大小溢出 */
			if (hashtable->batches[0].shared->ntuples + 1 >
				hashtable->nbuckets * NTUP_PER_BUCKET &&
				hashtable->nbuckets < (INT_MAX / 2) &&
				hashtable->nbuckets * 2 <=
				MaxAllocSize / sizeof(dsa_pointer_atomic))
			{
				fc_pstate->growth = PHJ_GROWTH_NEED_MORE_BUCKETS;
				LWLockRelease(&fc_pstate->lock);

				return NULL;
			}
		}
	}

	/* 我们被允许分配一个新的块。 */
	fc_chunk_shared = dsa_allocate(hashtable->area, fc_chunk_size);
	hashtable->batches[fc_curbatch].shared->size += fc_chunk_size;
	hashtable->batches[fc_curbatch].at_least_one_chunk = true;

	/* 设置块。 */
	fc_chunk = (HashMemoryChunk) dsa_get_address(hashtable->area, fc_chunk_shared);
	*fc_shared = fc_chunk_shared + HASH_CHUNK_HEADER_SIZE;
	fc_chunk->maxlen = fc_chunk_size - HASH_CHUNK_HEADER_SIZE;
	fc_chunk->used = fc_size;

	/*
	 * 推入块列表，以便如果我们需要增加桶或批次的数量（仅批次0），
	 * 并在稍后释放内存（所有批次）时可以找到它。
	 */
	fc_chunk->next.shared = hashtable->batches[fc_curbatch].shared->chunks;
	hashtable->batches[fc_curbatch].shared->chunks = fc_chunk_shared;

	if (fc_size <= HASH_CHUNK_THRESHOLD)
	{
		/*
		 * 使其成为当前块，以便我们可以在未来的调用中使用快速路径填充
		 * 剩余部分。
		 */
		hashtable->current_chunk = fc_chunk;
		hashtable->current_chunk_shared = fc_chunk_shared;
	}
	LWLockRelease(&fc_pstate->lock);

	Assert(HASH_CHUNK_DATA(fc_chunk) == dsa_get_address(hashtable->area, *fc_shared));
	fc_result = (HashJoinTuple) HASH_CHUNK_DATA(fc_chunk);

	return fc_result;
}

/*
 * 一个后端需要设置共享批次状态，包括tuplestores。
 * 其他后端将通过调用ExecParallelHashEnsureBatchAccessors()确保
 * 它们正确配置了访问器。
 */
static void fc_ExecParallelHashJoinSetUpBatches(HashJoinTable hashtable, int fc_nbatch)
{
	ParallelHashJoinState *fc_pstate = hashtable->parallel_state;
	ParallelHashJoinBatch *fc_batches;
	MemoryContext fc_oldcxt;
	int			fc_i;

	Assert(hashtable->batches == NULL);

	/* 分配空间。 */
	fc_pstate->batches =
		dsa_allocate0(hashtable->area,
					  EstimateParallelHashJoinBatch(hashtable) * fc_nbatch);
	fc_pstate->nbatch = fc_nbatch;
	fc_batches = dsa_get_address(hashtable->area, fc_pstate->batches);

	/* 使用哈希连接内存上下文。 */
	fc_oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);

	/* 分配该后端的访问器数组。 */
	hashtable->nbatch = fc_nbatch;
	hashtable->batches = (ParallelHashJoinBatchAccessor *)
		palloc0(sizeof(ParallelHashJoinBatchAccessor) * hashtable->nbatch);

	/* 设置共享状态、tuplestores和后端本地访问器。 */
	for (fc_i = 0; fc_i < hashtable->nbatch; ++fc_i)
	{
		ParallelHashJoinBatchAccessor *fc_accessor = &hashtable->batches[fc_i];
		ParallelHashJoinBatch *fc_shared = NthParallelHashJoinBatch(fc_batches, fc_i);
		char		fc_name[MAXPGPATH];

		/*
		 * 共享成员均已零初始化。我们只需要设置
		 * 屏障。
		 */
		BarrierInit(&fc_shared->batch_barrier, 0);
		if (fc_i == 0)
		{
			/* 批次0不需要加载。 */
			BarrierAttach(&fc_shared->batch_barrier);
			while (BarrierPhase(&fc_shared->batch_barrier) < PHJ_BATCH_PROBING)
				BarrierArriveAndWait(&fc_shared->batch_barrier, 0);
			BarrierDetach(&fc_shared->batch_barrier);
		}

		/* 初始化访问器状态。所有成员均已零初始化。 */
		fc_accessor->shared = fc_shared;

		/* 初始化共享tuplestores。 */
		snprintf(fc_name, sizeof(fc_name), "i%dof%d", fc_i, hashtable->nbatch);
		fc_accessor->inner_tuples =
			sts_initialize(ParallelHashJoinBatchInner(fc_shared),
						   fc_pstate->nparticipants,
						   ParallelWorkerNumber + 1,
						   sizeof(uint32),
						   SHARED_TUPLESTORE_SINGLE_PASS,
						   &fc_pstate->fileset,
						   fc_name);
		snprintf(fc_name, sizeof(fc_name), "o%dof%d", fc_i, hashtable->nbatch);
		fc_accessor->outer_tuples =
			sts_initialize(ParallelHashJoinBatchOuter(fc_shared,
													  fc_pstate->nparticipants),
						   fc_pstate->nparticipants,
						   ParallelWorkerNumber + 1,
						   sizeof(uint32),
						   SHARED_TUPLESTORE_SINGLE_PASS,
						   &fc_pstate->fileset,
						   fc_name);
	}

	MemoryContextSwitchTo(fc_oldcxt);
}

/*
 * 释放当前的ParallelHashJoinBatchAccessor对象集合。
 */
static void fc_ExecParallelHashCloseBatchAccessors(HashJoinTable hashtable)
{
	int			fc_i;

	for (fc_i = 0; fc_i < hashtable->nbatch; ++fc_i)
	{
		/* 确保没有文件保持打开状态。 */
		sts_end_write(hashtable->batches[fc_i].inner_tuples);
		sts_end_write(hashtable->batches[fc_i].outer_tuples);
		sts_end_parallel_scan(hashtable->batches[fc_i].inner_tuples);
		sts_end_parallel_scan(hashtable->batches[fc_i].outer_tuples);
	}
	pfree(hashtable->batches);
	hashtable->batches = NULL;
}

/*
 * 确保该后端对当前批次集合拥有最新的访问器。
 */
static void fc_ExecParallelHashEnsureBatchAccessors(HashJoinTable hashtable)
{
	ParallelHashJoinState *fc_pstate = hashtable->parallel_state;
	ParallelHashJoinBatch *fc_batches;
	MemoryContext fc_oldcxt;
	int			fc_i;

	if (hashtable->batches != NULL)
	{
		if (hashtable->nbatch == fc_pstate->nbatch)
			return;
		fc_ExecParallelHashCloseBatchAccessors(hashtable);
	}

	/*
	 * 我们绝对不应该看到批次跟踪数组被释放的状态，
	 * 因为如果我们在构建屏障达到PHJ_BUILD_DONE阶段时加入
	 * 我们应该提前放弃。
	 */
	Assert(DsaPointerIsValid(fc_pstate->batches));

	/* 使用哈希连接内存上下文。 */
	fc_oldcxt = MemoryContextSwitchTo(hashtable->hashCxt);

	/* 分配该后端的访问器数组。 */
	hashtable->nbatch = fc_pstate->nbatch;
	hashtable->batches = (ParallelHashJoinBatchAccessor *)
		palloc0(sizeof(ParallelHashJoinBatchAccessor) * hashtable->nbatch);

	/* 找到ParallelHashJoinBatch对象伪数组的基址。 */
	fc_batches = (ParallelHashJoinBatch *)
		dsa_get_address(hashtable->area, fc_pstate->batches);

	/* 设置访问器数组并附加到tuplestores。 */
	for (fc_i = 0; fc_i < hashtable->nbatch; ++fc_i)
	{
		ParallelHashJoinBatchAccessor *fc_accessor = &hashtable->batches[fc_i];
		ParallelHashJoinBatch *fc_shared = NthParallelHashJoinBatch(fc_batches, fc_i);

		fc_accessor->shared = fc_shared;
		fc_accessor->preallocated = 0;
		fc_accessor->done = false;
		fc_accessor->inner_tuples =
			sts_attach(ParallelHashJoinBatchInner(fc_shared),
					   ParallelWorkerNumber + 1,
					   &fc_pstate->fileset);
		fc_accessor->outer_tuples =
			sts_attach(ParallelHashJoinBatchOuter(fc_shared,
												  fc_pstate->nparticipants),
					   ParallelWorkerNumber + 1,
					   &fc_pstate->fileset);
	}

	MemoryContextSwitchTo(fc_oldcxt);
}

/*
 * 为给定批次分配一个空的共享内存哈希表。
 */
void ExecParallelHashTableAlloc(HashJoinTable hashtable, int fc_batchno)
{
	ParallelHashJoinBatch *fc_batch = hashtable->batches[fc_batchno].shared;
	dsa_pointer_atomic *fc_buckets;
	int			fc_nbuckets = hashtable->parallel_state->nbuckets;
	int			fc_i;

	fc_batch->buckets =
		dsa_allocate(hashtable->area, sizeof(dsa_pointer_atomic) * fc_nbuckets);
	fc_buckets = (dsa_pointer_atomic *)
		dsa_get_address(hashtable->area, fc_batch->buckets);
	for (fc_i = 0; fc_i < fc_nbuckets; ++fc_i)
		dsa_pointer_atomic_init(&fc_buckets[fc_i], InvalidDsaPointer);
}

/*
 * 如果我们当前附加到一个共享哈希连接批次，解除附加。如果我们
 * 是最后一个解除附加的，进行清理。
 */
void ExecHashTableDetachBatch(HashJoinTable hashtable)
{
	if (hashtable->parallel_state != NULL &&
		hashtable->curbatch >= 0)
	{
		int			fc_curbatch = hashtable->curbatch;
		ParallelHashJoinBatch *fc_batch = hashtable->batches[fc_curbatch].shared;

		/* 确保任何临时文件都已关闭。 */
		sts_end_parallel_scan(hashtable->batches[fc_curbatch].inner_tuples);
		sts_end_parallel_scan(hashtable->batches[fc_curbatch].outer_tuples);

		/* 从我们最后处理的批次中解除附加。 */
		if (BarrierArriveAndDetach(&fc_batch->batch_barrier))
		{
			/*
			 * 从技术上讲，我们不应该访问屏障，因为我们已经
			 * 不再附加，但由于在此之后它不可能移动，似乎安全
			 * 进行以下断言。
			 */
			Assert(BarrierPhase(&fc_batch->batch_barrier) == PHJ_BATCH_DONE);

			/* 释放共享块和桶。 */
			while (DsaPointerIsValid(fc_batch->chunks))
			{
				HashMemoryChunk fc_chunk =
				dsa_get_address(hashtable->area, fc_batch->chunks);
				dsa_pointer fc_next = fc_chunk->next.shared;

				dsa_free(hashtable->area, fc_batch->chunks);
				fc_batch->chunks = fc_next;
			}
			if (DsaPointerIsValid(fc_batch->buckets))
			{
				dsa_free(hashtable->area, fc_batch->buckets);
				fc_batch->buckets = InvalidDsaPointer;
			}
		}

		/*
		 * 跟踪我们附加过的最大的批次。尽管每个
		 * 后端可能会看到不同的批次子集，explain.c将
		 * 扫描所有后端的结果以找到最大值。
		 */
		hashtable->spacePeak =
			Max(hashtable->spacePeak,
				fc_batch->size + sizeof(dsa_pointer_atomic) * hashtable->nbuckets);

		/* 记住我们未附加到批次上。 */
		hashtable->curbatch = -1;
	}
}

/*
 * 从所有共享资源中解除附加。如果我们是最后一个解除附加的，进行清理。
 */
void ExecHashTableDetach(HashJoinTable hashtable)
{
	ParallelHashJoinState *fc_pstate = hashtable->parallel_state;

	/*
	 * 如果我们参与并行查询，我们必须已经到达PHJ_BUILD_RUNNING，或者
	 * 加入得太晚而处于PHJ_BUILD_DONE。
	 */
	Assert(!fc_pstate ||
		   BarrierPhase(&fc_pstate->build_barrier) >= PHJ_BUILD_RUNNING);

	if (fc_pstate && BarrierPhase(&fc_pstate->build_barrier) == PHJ_BUILD_RUNNING)
	{
		int			fc_i;

		/* 确保任何临时文件都已关闭。 */
		if (hashtable->batches)
		{
			for (fc_i = 0; fc_i < hashtable->nbatch; ++fc_i)
			{
				sts_end_write(hashtable->batches[fc_i].inner_tuples);
				sts_end_write(hashtable->batches[fc_i].outer_tuples);
				sts_end_parallel_scan(hashtable->batches[fc_i].inner_tuples);
				sts_end_parallel_scan(hashtable->batches[fc_i].outer_tuples);
			}
		}

		/* 如果我们是最后一个解除附加的，清理共享内存。 */
		if (BarrierArriveAndDetach(&fc_pstate->build_barrier))
		{
			/*
			 * 迟到的连接进程将看到此状态并立即放弃。
			 */
			Assert(BarrierPhase(&fc_pstate->build_barrier) == PHJ_BUILD_DONE);

			if (DsaPointerIsValid(fc_pstate->batches))
			{
				dsa_free(hashtable->area, fc_pstate->batches);
				fc_pstate->batches = InvalidDsaPointer;
			}
		}
	}
	hashtable->parallel_state = NULL;
}

/*
 * 在给定的桶中获取第一个元组，桶由编号识别。
 */
static inline HashJoinTuple
fc_ExecParallelHashFirstTuple(HashJoinTable hashtable, int fc_bucketno)
{
	HashJoinTuple fc_tuple;
	dsa_pointer fc_p;

	Assert(hashtable->parallel_state);
	fc_p = dsa_pointer_atomic_read(&hashtable->buckets.shared[fc_bucketno]);
	fc_tuple = (HashJoinTuple) dsa_get_address(hashtable->area, fc_p);

	return fc_tuple;
}

/*
 * 获取与'tuple'相同桶中的下一个元组。
 */
static inline HashJoinTuple
fc_ExecParallelHashNextTuple(HashJoinTable hashtable, HashJoinTuple fc_tuple)
{
	HashJoinTuple fc_next;

	Assert(hashtable->parallel_state);
	fc_next = (HashJoinTuple) dsa_get_address(hashtable->area, fc_tuple->next.shared);

	return fc_next;
}

/*
 * 原子性地在DSA内存中将一个元组插入元组链的前面。
 */
static inline void fc_ExecParallelHashPushTuple(dsa_pointer_atomic *fc_head,
						  HashJoinTuple fc_tuple,
						  dsa_pointer fc_tuple_shared)
{
	for (;;)
	{
		fc_tuple->next.shared = dsa_pointer_atomic_read(fc_head);
		if (dsa_pointer_atomic_compare_exchange(fc_head,
												&fc_tuple->next.shared,
												fc_tuple_shared))
			break;
	}
}

/*
 * 准备处理给定的批次。
 */
void ExecParallelHashTableSetCurrentBatch(HashJoinTable hashtable, int fc_batchno)
{
	Assert(hashtable->batches[fc_batchno].shared->buckets != InvalidDsaPointer);

	hashtable->curbatch = fc_batchno;
	hashtable->buckets.shared = (dsa_pointer_atomic *)
		dsa_get_address(hashtable->area,
						hashtable->batches[fc_batchno].shared->buckets);
	hashtable->nbuckets = hashtable->parallel_state->nbuckets;
	hashtable->log2_nbuckets = my_log2(hashtable->nbuckets);
	hashtable->current_chunk = NULL;
	hashtable->current_chunk_shared = InvalidDsaPointer;
	hashtable->batches[fc_batchno].at_least_one_chunk = false;
}

/*
 * 从并行处理中的块队列中获取下一个可用的块。如果没有剩余的，则返回NULL。否则返回指向块的指针，并将*shared设置为指向该块的DSA指针。
 */
static HashMemoryChunk
fc_ExecParallelHashPopChunkQueue(HashJoinTable hashtable, dsa_pointer *fc_shared)
{
	ParallelHashJoinState *fc_pstate = hashtable->parallel_state;
	HashMemoryChunk fc_chunk;

	LWLockAcquire(&fc_pstate->lock, LW_EXCLUSIVE);
	if (DsaPointerIsValid(fc_pstate->chunk_work_queue))
	{
		*fc_shared = fc_pstate->chunk_work_queue;
		fc_chunk = (HashMemoryChunk)
			dsa_get_address(hashtable->area, *fc_shared);
		fc_pstate->chunk_work_queue = fc_chunk->next.shared;
	}
	else
		fc_chunk = NULL;
	LWLockRelease(&fc_pstate->lock);

	return fc_chunk;
}

/*
 * 为给定的内部批次增加在此后端预分配的空间，至少增加给定的数量。这使我们能够跟踪在重新加载时给定的批次是否适合内存。如果需要，还增加批次或桶的数量。
 *
 * 通过模拟块将如何分配给工作者来维持一个持续的空间使用估计。当我们将批次重新加载到内存时，这样可以估算所需空间。它并不完全准确，因为ExecParallelHashTupleAlloc()会以不同的方式将元组打包到内存块中，但应该相当接近。由于所有工作者在哈希期间都会一起预分配，因此它倾向于每个工作者多估计一小部分块，但是如果有足够的工作者分配，工人通常会单独重新加载批次，从而留下更少的部分填充块。该效应被nparticipants所限制。
 *
 * 如果批次或桶的数量发生变化，返回false，调用者应重新考虑给定元组现在属于哪个批次并再次调用。
 */
static bool fc_ExecParallelHashTuplePrealloc(HashJoinTable hashtable, int fc_batchno, size_t fc_size)
{
	ParallelHashJoinState *fc_pstate = hashtable->parallel_state;
	ParallelHashJoinBatchAccessor *fc_batch = &hashtable->batches[fc_batchno];
	size_t		fc_want = Max(fc_size, HASH_CHUNK_SIZE - HASH_CHUNK_HEADER_SIZE);

	Assert(fc_batchno > 0);
	Assert(fc_batchno < hashtable->nbatch);
	Assert(fc_size == MAXALIGN(fc_size));

	LWLockAcquire(&fc_pstate->lock, LW_EXCLUSIVE);

	/* 还有其他参与者要求我们帮助增长吗？ */
	if (fc_pstate->growth == PHJ_GROWTH_NEED_MORE_BATCHES ||
		fc_pstate->growth == PHJ_GROWTH_NEED_MORE_BUCKETS)
	{
		ParallelHashGrowth fc_growth = fc_pstate->growth;

		LWLockRelease(&fc_pstate->lock);
		if (fc_growth == PHJ_GROWTH_NEED_MORE_BATCHES)
			fc_ExecParallelHashIncreaseNumBatches(hashtable);
		else if (fc_growth == PHJ_GROWTH_NEED_MORE_BUCKETS)
			fc_ExecParallelHashIncreaseNumBuckets(hashtable);

		return false;
	}

	if (fc_pstate->growth != PHJ_GROWTH_DISABLED &&
		fc_batch->at_least_one_chunk &&
		(fc_batch->shared->estimated_size + fc_want + HASH_CHUNK_HEADER_SIZE
		 > fc_pstate->space_allowed))
	{
		/*
		 * 我们已经确定如果将此批次加载到内存中，则会超出空间预算。指示所有参与者帮助重新分区。
		 */
		fc_batch->shared->space_exhausted = true;
		fc_pstate->growth = PHJ_GROWTH_NEED_MORE_BATCHES;
		LWLockRelease(&fc_pstate->lock);

		return false;
	}

	fc_batch->at_least_one_chunk = true;
	fc_batch->shared->estimated_size += fc_want + HASH_CHUNK_HEADER_SIZE;
	fc_batch->preallocated = fc_want;
	LWLockRelease(&fc_pstate->lock);

	return true;
}

/*
 * 计算Hash和类似计划类型可以使用的内存限制。这是work_mem乘以hash_mem_multiplier，并以字节为单位表示。
 *
 * 为规划器以及其他类似哈希的执行节点导出。这是一个相当随机的地方，但没有更好的地方。
 */
size_t get_hash_memory_limit(void)
{
	double		fc_mem_limit;

	/* 在双精度算术中执行初步计算 */
	fc_mem_limit = (double) work_mem * hash_mem_multiplier * 1024.0;

	/* 在大小不合适时进行限制 */
	fc_mem_limit = Min(fc_mem_limit, (double) SIZE_MAX);

	return (size_t) fc_mem_limit;
}
